Merge #20685: Add I2P support using I2P SAM

a701fcf01f3ea9a12e869bfa52321302cf68351c net: Do not skip the I2P network from GetNetworkNames() (Vasil Dimov)
0181e244394bd9e68e9f0d44704e7b0fd12a6b1f net: recognize I2P from ParseNetwork() so that -onlynet=i2p works (Vasil Dimov)
b905363fa8b0bb03fe34b53b5410880f42e0af39 net: accept incoming I2P connections from CConnman (Vasil Dimov)
0635233a1e7e8c303073430092afd3e0fb0d927b net: make outgoing I2P connections from CConnman (Vasil Dimov)
9559bd1404fbf74b0d09fe9019a9305cb4e151ce net: add I2P to the reachability map (Vasil Dimov)
76c35c60f338937071bcfad4211ef7254d3830ec init: introduce I2P connectivity options (Vasil Dimov)
c22daa2ecff1acd25426cd46f98f2587d1d324c3 net: implement the necessary parts of the I2P SAM protocol (Vasil Dimov)
5bac7e45e1d3a07115b5ff002d988438fcc92a53 net: extend Sock with a method to check whether connected (Vasil Dimov)
42c779f503eb8437b6232773a4a2472306cc9f3d net: extend Sock with methods for robust send & read until terminator (Vasil Dimov)
ea1845315a109eb105113cb5fbb6f869e1cf010c net: extend Sock::Wait() to report a timeout (Vasil Dimov)
78fdfbea666201b25919dd67454eb04d6a34326f net: dedup MSG_NOSIGNAL and MSG_DONTWAIT definitions (Vasil Dimov)
34bcfab562bac9887ca9c3831cf4fd0ee7f98149 net: move the constant maxWait out of InterruptibleRecv() (Vasil Dimov)
cff65c4a270887ec171293409ab84f5d0d0be7fc net: extend CNetAddr::SetSpecial() to support I2P (Vasil Dimov)
f6c267db3be2d7077fd2bdbd34860eba838dea99 net: avoid unnecessary GetBindAddress() call (Vasil Dimov)
7c224fdac4699a2c4953b33ab423f9cddbf95cf7 net: isolate the protocol-agnostic part of CConnman::AcceptConnection() (Vasil Dimov)
1f75a653dd3b24ba2e4383bf951a6e5a3d5ccbcf net: get the bind address earlier in CConnman::AcceptConnection() (Vasil Dimov)
25605895afe84b1765dd9da9240af22f99489df7 net: check for invalid socket earlier in CConnman::AcceptConnection() (Vasil Dimov)
545bc5f81d60fa6ff7c5cc43a2e9eef82f911466 util: fix WriteBinaryFile() claiming success even if error occurred (Vasil Dimov)
8b6e4b3b23027da263d257b342f5d9a53e4032d5 util: fix ReadBinaryFile() returning partial contents (Vasil Dimov)
4cba2fdafa483cbdb70f581174138ec253c80d48 util: extract {Read,Write}BinaryFile() to its own files (Vasil Dimov)

Pull request description:

  Add I2P support by using the [I2P SAM](https://geti2p.net/en/docs/api/samv3) protocol. Unlike Tor, for incoming connections we get the I2P address of the peer (and they also receive ours when we are the connection initiator).

  Two new options are added:

  ```
    -i2psam=<ip:port>
         I2P SAM proxy to reach I2P peers and accept I2P connections (default:
         none)

    -i2pacceptincoming
         If set and -i2psam is also set then incoming I2P connections are
         accepted via the SAM proxy. If this is not set but -i2psam is set
         then only outgoing connections will be made to the I2P network.
         Ignored if -i2psam is not set. Notice that listening for incoming
         I2P connections is done through the SAM proxy, not by binding to
         a local address and port (default: true)
  ```

  # Overview of the changes

  ## Make `ReadBinary()` and `WriteBinary()` reusable

  We would need to dump the I2P private key to a file and read it back later. Move those two functions out of `torcontrol.cpp`.

  ```
  util: extract {Read,Write}BinaryFile() to its own files
  util: fix ReadBinaryFile() returning partial contents
  util: fix WriteBinaryFile() claiming success even if error occurred
  ```

  ## Split `CConnman::AcceptConnection()`

  Most of `CConnman::AcceptConnection()` is agnostic of how the socket was accepted. The other part of it deals with the details of the `accept(2)` system call. Split those so that the protocol-agnostic part can be reused if we accept a socket by other means.

  ```
  net: check for invalid socket earlier in CConnman::AcceptConnection()
  net: get the bind address earlier in CConnman::AcceptConnection()
  net: isolate the protocol-agnostic part of CConnman::AcceptConnection()
  net: avoid unnecessary GetBindAddress() call
  ```

  ## Implement the I2P [SAM](https://geti2p.net/en/docs/api/samv3) protocol (not all of it)

  Just the parts that would enable us to make outgoing and accept incoming I2P connections.

  ```
  net: extend CNetAddr::SetSpecial() to support I2P
  net: move the constant maxWait out of InterruptibleRecv()
  net: dedup MSG_NOSIGNAL and MSG_DONTWAIT definitions
  net: extend Sock::Wait() to report a timeout
  net: extend Sock with methods for robust send & read until terminator
  net: extend Sock with a method to check whether connected
  net: implement the necessary parts of the I2P SAM protocol
  ```

  ## Use I2P SAM to connect to and accept connections from I2P peers

  Profit from all of the preceding commits.

  ```
  init: introduce I2P connectivity options
  net: add I2P to the reachability map
  net: make outgoing I2P connections from CConnman
  net: accept incoming I2P connections from CConnman
  net: recognize I2P from ParseNetwork() so that -onlynet=i2p works
  net: Do not skip the I2P network from GetNetworkNames()
  ```

ACKs for top commit:
  laanwj:
    re-ACK a701fcf01f3ea9a12e869bfa52321302cf68351c
  jonatack:
    re-ACK a701fcf01f3ea9a12e869bfa52321302cf68351c reviewed diff per `git range-diff ad89812 2a7bb34 a701fcf`, debug built and launched bitcoind with i2pd v2.35 running a dual I2P+Torv3 service with the I2P config settings listed below (did not test `onlynet=i2p`); operation appears nominal (same as it has been these past weeks), and tested the bitcoind help outputs grepping for `-i i2p` and the rpc getpeerinfo and getnetworkinfo helps

Tree-SHA512: de42090c9c0bf23b43b5839f5b4fc4b3a2657bde1e45c796b5f3c7bf83cb8ec6ca4278f8a89e45108ece92f9b573cafea3b42a06bc09076b40a196c909b6610e

19 files changed, 1267 insertions, 104 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index eae226b1d4..1c6d97b714 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -148,6 +148,7 @@ BITCOIN_CORE_H = \
   fs.h \
   httprpc.h \
   httpserver.h \
+  i2p.h \
   index/base.h \
   index/blockfilterindex.h \
   index/disktxpos.h \
@@ -242,6 +243,7 @@ BITCOIN_CORE_H = \
   util/message.h \
   util/moneystr.h \
   util/rbf.h \
+  util/readwritefile.h \
   util/ref.h \
   util/settings.h \
   util/sock.h \
@@ -314,6 +316,7 @@ libbitcoin_server_a_SOURCES = \
   flatfile.cpp \
   httprpc.cpp \
   httpserver.cpp \
+  i2p.cpp \
   index/base.cpp \
   index/blockfilterindex.cpp \
   index/txindex.cpp \
@@ -572,6 +575,7 @@ libbitcoin_util_a_SOURCES = \
   util/message.cpp \
   util/moneystr.cpp \
   util/rbf.cpp \
+  util/readwritefile.cpp \
   util/settings.cpp \
   util/threadnames.cpp \
   util/spanparsing.cpp \
diff --git a/src/compat.h b/src/compat.h
index dad14748a2..3449bc2661 100644
--- a/src/compat.h
+++ b/src/compat.h
@@ -44,6 +44,7 @@ typedef unsigned int SOCKET;
 #define WSAEINVAL           EINVAL
 #define WSAEALREADY         EALREADY
 #define WSAEWOULDBLOCK      EWOULDBLOCK
+#define WSAEAGAIN           EAGAIN
 #define WSAEMSGSIZE         EMSGSIZE
 #define WSAEINTR            EINTR
 #define WSAEINPROGRESS      EINPROGRESS
@@ -51,6 +52,14 @@ typedef unsigned int SOCKET;
 #define WSAENOTSOCK         EBADF
 #define INVALID_SOCKET      (SOCKET)(~0)
 #define SOCKET_ERROR        -1
+#else
+#ifndef WSAEAGAIN
+#ifdef EAGAIN
+#define WSAEAGAIN EAGAIN
+#else
+#define WSAEAGAIN WSAEWOULDBLOCK
+#endif
+#endif
 #endif
 
 #ifdef WIN32
@@ -96,4 +105,14 @@ bool static inline IsSelectableSocket(const SOCKET& s) {
 #endif
 }
 
+// MSG_NOSIGNAL is not available on some platforms, if it doesn't exist define it as 0
+#if !defined(MSG_NOSIGNAL)
+#define MSG_NOSIGNAL 0
+#endif
+
+// MSG_DONTWAIT is not available on some platforms, if it doesn't exist define it as 0
+#if !defined(MSG_DONTWAIT)
+#define MSG_DONTWAIT 0
+#endif
+
 #endif // BITCOIN_COMPAT_H
diff --git a/src/i2p.cpp b/src/i2p.cpp
new file mode 100644
index 0000000000..42270deaeb
--- /dev/null
+++ b/src/i2p.cpp
@@ -0,0 +1,407 @@
+// Copyright (c) 2020-2020 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <chainparams.h>
+#include <compat.h>
+#include <compat/endian.h>
+#include <crypto/sha256.h>
+#include <fs.h>
+#include <i2p.h>
+#include <logging.h>
+#include <netaddress.h>
+#include <netbase.h>
+#include <random.h>
+#include <util/strencodings.h>
+#include <tinyformat.h>
+#include <util/readwritefile.h>
+#include <util/sock.h>
+#include <util/spanparsing.h>
+#include <util/system.h>
+
+#include <chrono>
+#include <stdexcept>
+#include <string>
+
+namespace i2p {
+
+/**
+ * Swap Standard Base64 <-> I2P Base64.
+ * Standard Base64 uses `+` and `/` as last two characters of its alphabet.
+ * I2P Base64 uses `-` and `~` respectively.
+ * So it is easy to detect in which one is the input and convert to the other.
+ * @param[in] from Input to convert.
+ * @return converted `from`
+ */
+static std::string SwapBase64(const std::string& from)
+{
+    std::string to;
+    to.resize(from.size());
+    for (size_t i = 0; i < from.size(); ++i) {
+        switch (from[i]) {
+        case '-':
+            to[i] = '+';
+            break;
+        case '~':
+            to[i] = '/';
+            break;
+        case '+':
+            to[i] = '-';
+            break;
+        case '/':
+            to[i] = '~';
+            break;
+        default:
+            to[i] = from[i];
+            break;
+        }
+    }
+    return to;
+}
+
+/**
+ * Decode an I2P-style Base64 string.
+ * @param[in] i2p_b64 I2P-style Base64 string.
+ * @return decoded `i2p_b64`
+ * @throw std::runtime_error if decoding fails
+ */
+static Binary DecodeI2PBase64(const std::string& i2p_b64)
+{
+    const std::string& std_b64 = SwapBase64(i2p_b64);
+    bool invalid;
+    Binary decoded = DecodeBase64(std_b64.c_str(), &invalid);
+    if (invalid) {
+        throw std::runtime_error(strprintf("Cannot decode Base64: \"%s\"", i2p_b64));
+    }
+    return decoded;
+}
+
+/**
+ * Derive the .b32.i2p address of an I2P destination (binary).
+ * @param[in] dest I2P destination.
+ * @return the address that corresponds to `dest`
+ * @throw std::runtime_error if conversion fails
+ */
+static CNetAddr DestBinToAddr(const Binary& dest)
+{
+    CSHA256 hasher;
+    hasher.Write(dest.data(), dest.size());
+    unsigned char hash[CSHA256::OUTPUT_SIZE];
+    hasher.Finalize(hash);
+
+    CNetAddr addr;
+    const std::string addr_str = EncodeBase32(hash, false) + ".b32.i2p";
+    if (!addr.SetSpecial(addr_str)) {
+        throw std::runtime_error(strprintf("Cannot parse I2P address: \"%s\"", addr_str));
+    }
+
+    return addr;
+}
+
+/**
+ * Derive the .b32.i2p address of an I2P destination (I2P-style Base64).
+ * @param[in] dest I2P destination.
+ * @return the address that corresponds to `dest`
+ * @throw std::runtime_error if conversion fails
+ */
+static CNetAddr DestB64ToAddr(const std::string& dest)
+{
+    const Binary& decoded = DecodeI2PBase64(dest);
+    return DestBinToAddr(decoded);
+}
+
+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)
+{
+}
+
+Session::~Session()
+{
+    LOCK(m_mutex);
+    Disconnect();
+}
+
+bool Session::Listen(Connection& conn)
+{
+    try {
+        LOCK(m_mutex);
+        CreateIfNotCreatedAlready();
+        conn.me = m_my_addr;
+        conn.sock = StreamAccept();
+        return true;
+    } catch (const std::runtime_error& e) {
+        Log("Error listening: %s", e.what());
+        CheckControlSock();
+    }
+    return false;
+}
+
+bool Session::Accept(Connection& conn)
+{
+    try {
+        while (!*m_interrupt) {
+            Sock::Event 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);
+
+            conn.peer = CService(DestB64ToAddr(peer_dest), Params().GetDefaultPort());
+
+            return true;
+        }
+    } catch (const std::runtime_error& e) {
+        Log("Error accepting: %s", e.what());
+        CheckControlSock();
+    }
+    return false;
+}
+
+bool Session::Connect(const CService& to, Connection& conn, bool& proxy_error)
+{
+    proxy_error = true;
+
+    std::string session_id;
+    Sock sock;
+    conn.peer = to;
+
+    try {
+        {
+            LOCK(m_mutex);
+            CreateIfNotCreatedAlready();
+            session_id = m_session_id;
+            conn.me = m_my_addr;
+            sock = Hello();
+        }
+
+        const Reply& lookup_reply =
+            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),
+            false);
+
+        const std::string& result = connect_reply.Get("RESULT");
+
+        if (result == "OK") {
+            conn.sock = std::move(sock);
+            return true;
+        }
+
+        if (result == "INVALID_ID") {
+            LOCK(m_mutex);
+            Disconnect();
+            throw std::runtime_error("Invalid session id");
+        }
+
+        if (result == "CANT_REACH_PEER" || result == "TIMEOUT") {
+            proxy_error = false;
+        }
+
+        throw std::runtime_error(strprintf("\"%s\"", connect_reply.full));
+    } catch (const std::runtime_error& e) {
+        Log("Error connecting to %s: %s", to.ToString(), e.what());
+        CheckControlSock();
+        return false;
+    }
+}
+
+// Private methods
+
+std::string Session::Reply::Get(const std::string& key) const
+{
+    const auto& pos = keys.find(key);
+    if (pos == keys.end() || !pos->second.has_value()) {
+        throw std::runtime_error(
+            strprintf("Missing %s= in the reply to \"%s\": \"%s\"", key, request, full));
+    }
+    return pos->second.value();
+}
+
+template <typename... Args>
+void Session::Log(const std::string& fmt, const Args&... args) const
+{
+    LogPrint(BCLog::I2P, "I2P: %s\n", tfm::format(fmt, args...));
+}
+
+Session::Reply Session::SendRequestAndGetReply(const Sock& sock,
+                                               const std::string& request,
+                                               bool check_result_ok) const
+{
+    sock.SendComplete(request + "\n", MAX_WAIT_FOR_IO, *m_interrupt);
+
+    Reply reply;
+
+    // Don't log the full "SESSION CREATE ..." because it contains our private key.
+    reply.request = request.substr(0, 14) == "SESSION CREATE" ? "SESSION CREATE ..." : request;
+
+    // It could take a few minutes for the I2P router to reply as it is querying the I2P network
+    // (when doing name lookup, for example). Notice: `RecvUntilTerminator()` is checking
+    // `m_interrupt` more often, so we would not be stuck here for long if `m_interrupt` is
+    // signaled.
+    static constexpr auto recv_timeout = 3min;
+
+    reply.full = sock.RecvUntilTerminator('\n', recv_timeout, *m_interrupt);
+
+    for (const auto& kv : spanparsing::Split(reply.full, ' ')) {
+        const auto& pos = std::find(kv.begin(), kv.end(), '=');
+        if (pos != kv.end()) {
+            reply.keys.emplace(std::string{kv.begin(), pos}, std::string{pos + 1, kv.end()});
+        } else {
+            reply.keys.emplace(std::string{kv.begin(), kv.end()}, std::nullopt);
+        }
+    }
+
+    if (check_result_ok && reply.Get("RESULT") != "OK") {
+        throw std::runtime_error(
+            strprintf("Unexpected reply to \"%s\": \"%s\"", request, reply.full));
+    }
+
+    return reply;
+}
+
+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)) {
+        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);
+}
+
+void Session::CheckControlSock()
+{
+    LOCK(m_mutex);
+
+    std::string errmsg;
+    if (!m_control_sock.IsConnected(errmsg)) {
+        Log("Control socket error: %s", errmsg);
+        Disconnect();
+    }
+}
+
+void Session::DestGenerate(const Sock& sock)
+{
+    // https://geti2p.net/spec/common-structures#key-certificates
+    // "7" or "EdDSA_SHA512_Ed25519" - "Recent Router Identities and Destinations".
+    // Use "7" because i2pd <2.24.0 does not recognize the textual form.
+    const Reply& reply = SendRequestAndGetReply(sock, "DEST GENERATE SIGNATURE_TYPE=7", false);
+
+    m_private_key = DecodeI2PBase64(reply.Get("PRIV"));
+}
+
+void Session::GenerateAndSavePrivateKey(const Sock& sock)
+{
+    DestGenerate(sock);
+
+    // umask is set to 077 in init.cpp, which is ok (unless -sysperms is given)
+    if (!WriteBinaryFile(m_private_key_file,
+                         std::string(m_private_key.begin(), m_private_key.end()))) {
+        throw std::runtime_error(
+            strprintf("Cannot save I2P private key to %s", m_private_key_file));
+    }
+}
+
+Binary Session::MyDestination() const
+{
+    // From https://geti2p.net/spec/common-structures#destination:
+    // "They are 387 bytes plus the certificate length specified at bytes 385-386, which may be
+    // non-zero"
+    static constexpr size_t DEST_LEN_BASE = 387;
+    static constexpr size_t CERT_LEN_POS = 385;
+
+    uint16_t cert_len;
+    memcpy(&cert_len, &m_private_key.at(CERT_LEN_POS), sizeof(cert_len));
+    cert_len = be16toh(cert_len);
+
+    const size_t dest_len = DEST_LEN_BASE + cert_len;
+
+    return Binary{m_private_key.begin(), m_private_key.begin() + dest_len};
+}
+
+void Session::CreateIfNotCreatedAlready()
+{
+    std::string errmsg;
+    if (m_control_sock.IsConnected(errmsg)) {
+        return;
+    }
+
+    Log("Creating SAM session with %s", m_control_host.ToString());
+
+    Sock 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);
+    }
+
+    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));
+
+    m_my_addr = CService(DestBinToAddr(MyDestination()), Params().GetDefaultPort());
+    m_session_id = session_id;
+    m_control_sock = std::move(sock);
+
+    LogPrintf("I2P: SAM session created: session id=%s, my address=%s\n", m_session_id,
+              m_my_addr.ToString());
+}
+
+Sock Session::StreamAccept()
+{
+    Sock sock = Hello();
+
+    const Reply& reply = SendRequestAndGetReply(
+        sock, strprintf("STREAM ACCEPT ID=%s SILENT=false", m_session_id), false);
+
+    const std::string& result = reply.Get("RESULT");
+
+    if (result == "OK") {
+        return sock;
+    }
+
+    if (result == "INVALID_ID") {
+        // If our session id is invalid, then force session re-creation on next usage.
+        Disconnect();
+    }
+
+    throw std::runtime_error(strprintf("\"%s\"", reply.full));
+}
+
+void Session::Disconnect()
+{
+    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_session_id.clear();
+}
+} // namespace sam
+} // namespace i2p
diff --git a/src/i2p.h b/src/i2p.h
new file mode 100644
index 0000000000..8fafe0a4d0
--- /dev/null
+++ b/src/i2p.h
@@ -0,0 +1,260 @@
+// Copyright (c) 2020-2020 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_I2P_H
+#define BITCOIN_I2P_H
+
+#include <compat.h>
+#include <fs.h>
+#include <netaddress.h>
+#include <sync.h>
+#include <threadinterrupt.h>
+#include <util/sock.h>
+
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+namespace i2p {
+
+/**
+ * Binary data.
+ */
+using Binary = std::vector<uint8_t>;
+
+/**
+ * An established connection with another peer.
+ */
+struct Connection {
+    /** Connected socket. */
+    Sock sock;
+
+    /** Our I2P address. */
+    CService me;
+
+    /** The peer's I2P address. */
+    CService peer;
+};
+
+namespace sam {
+
+/**
+ * I2P SAM session.
+ */
+class Session
+{
+public:
+    /**
+     * Construct a session. This will not initiate any IO, the session will be lazily created
+     * later when first used.
+     * @param[in] private_key_file Path to a private key file. If the file does not exist then the
+     * private key will be generated and saved into the file.
+     * @param[in] control_host Location of the SAM proxy.
+     * @param[in,out] interrupt If this is signaled then all operations are canceled as soon as
+     * possible and executing methods throw an exception. Notice: only a pointer to the
+     * `CThreadInterrupt` object is saved, so it must not be destroyed earlier than this
+     * `Session` object.
+     */
+    Session(const fs::path& private_key_file,
+            const CService& control_host,
+            CThreadInterrupt* interrupt);
+
+    /**
+     * Destroy the session, closing the internally used sockets. The sockets that have been
+     * returned by `Accept()` or `Connect()` will not be closed, but they will be closed by
+     * the SAM proxy because the session is destroyed. So they will return an error next time
+     * we try to read or write to them.
+     */
+    ~Session();
+
+    /**
+     * Start listening for an incoming connection.
+     * @param[out] conn Upon successful completion the `sock` and `me` members will be set
+     * to the listening socket and address.
+     * @return true on success
+     */
+    bool Listen(Connection& conn);
+
+    /**
+     * Wait for and accept a new incoming connection.
+     * @param[in,out] conn The `sock` member is used for waiting and accepting. Upon successful
+     * completion the `peer` member will be set to the address of the incoming peer.
+     * @return true on success
+     */
+    bool Accept(Connection& conn);
+
+    /**
+     * Connect to an I2P peer.
+     * @param[in] to Peer to connect to.
+     * @param[out] conn Established connection. Only set if `true` is returned.
+     * @param[out] proxy_error If an error occurs due to proxy or general network failure, then
+     * this is set to `true`. If an error occurs due to unreachable peer (likely peer is down), then
+     * it is set to `false`. Only set if `false` is returned.
+     * @return true on success
+     */
+    bool Connect(const CService& to, Connection& conn, bool& proxy_error);
+
+private:
+    /**
+     * A reply from the SAM proxy.
+     */
+    struct Reply {
+        /**
+         * Full, unparsed reply.
+         */
+        std::string full;
+
+        /**
+         * Request, used for detailed error reporting.
+         */
+        std::string request;
+
+        /**
+         * A map of keywords from the parsed reply.
+         * For example, if the reply is "A=X B C=YZ", then the map will be
+         * keys["A"] == "X"
+         * keys["B"] == (empty std::optional)
+         * keys["C"] == "YZ"
+         */
+        std::unordered_map<std::string, std::optional<std::string>> keys;
+
+        /**
+         * Get the value of a given key.
+         * For example if the reply is "A=X B" then:
+         * Value("A") -> "X"
+         * Value("B") -> throws
+         * Value("C") -> throws
+         * @param[in] key Key whose value to retrieve
+         * @returns the key's value
+         * @throws std::runtime_error if the key is not present or if it has no value
+         */
+        std::string Get(const std::string& key) const;
+    };
+
+    /**
+     * Log a message in the `BCLog::I2P` category.
+     * @param[in] fmt printf(3)-like format string.
+     * @param[in] args printf(3)-like arguments that correspond to `fmt`.
+     */
+    template <typename... Args>
+    void Log(const std::string& fmt, const Args&... args) const;
+
+    /**
+     * Send request and get a reply from the SAM proxy.
+     * @param[in] sock A socket that is connected to the SAM proxy.
+     * @param[in] request Raw request to send, a newline terminator is appended to it.
+     * @param[in] check_result_ok If true then after receiving the reply a check is made
+     * whether it contains "RESULT=OK" and an exception is thrown if it does not.
+     * @throws std::runtime_error if an error occurs
+     */
+    Reply SendRequestAndGetReply(const Sock& sock,
+                                 const std::string& request,
+                                 bool check_result_ok = true) const;
+
+    /**
+     * Open a new connection to the SAM proxy.
+     * @return a connected socket
+     * @throws std::runtime_error if an error occurs
+     */
+    Sock Hello() const EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+
+    /**
+     * Check the control socket for errors and possibly disconnect.
+     */
+    void CheckControlSock();
+
+    /**
+     * Generate a new destination with the SAM proxy and set `m_private_key` to it.
+     * @param[in] sock Socket to use for talking to the SAM proxy.
+     * @throws std::runtime_error if an error occurs
+     */
+    void DestGenerate(const Sock& sock) EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+
+    /**
+     * Generate a new destination with the SAM proxy, set `m_private_key` to it and save
+     * it on disk to `m_private_key_file`.
+     * @param[in] sock Socket to use for talking to the SAM proxy.
+     * @throws std::runtime_error if an error occurs
+     */
+    void GenerateAndSavePrivateKey(const Sock& sock) EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+
+    /**
+     * Derive own destination from `m_private_key`.
+     * @see https://geti2p.net/spec/common-structures#destination
+     * @return an I2P destination
+     */
+    Binary MyDestination() const EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+
+    /**
+     * Create the session if not already created. Reads the private key file and connects to the
+     * SAM proxy.
+     * @throws std::runtime_error if an error occurs
+     */
+    void CreateIfNotCreatedAlready() EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+
+    /**
+     * 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.
+     * @throws std::runtime_error if an error occurs
+     */
+    Sock StreamAccept() EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+
+    /**
+     * Destroy the session, closing the internally used sockets.
+     */
+    void Disconnect() EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+
+    /**
+     * The name of the file where this peer's private key is stored (in binary).
+     */
+    const fs::path m_private_key_file;
+
+    /**
+     * The host and port of the SAM control service.
+     */
+    const CService m_control_host;
+
+    /**
+     * Cease network activity when this is signaled.
+     */
+    CThreadInterrupt* const m_interrupt;
+
+    /**
+     * Mutex protecting the members that can be concurrently accessed.
+     */
+    mutable Mutex m_mutex;
+
+    /**
+     * The private key of this peer.
+     * @see The reply to the "DEST GENERATE" command in https://geti2p.net/en/docs/api/samv3
+     */
+    Binary m_private_key GUARDED_BY(m_mutex);
+
+    /**
+     * SAM control socket.
+     * Used to connect to the I2P SAM service and create a session
+     * ("SESSION CREATE"). With the established session id we later open
+     * other connections to the SAM service to accept incoming I2P
+     * connections and make outgoing ones.
+     * See https://geti2p.net/en/docs/api/samv3
+     */
+    Sock m_control_sock GUARDED_BY(m_mutex);
+
+    /**
+     * Our .b32.i2p address.
+     * Derived from `m_private_key`.
+     */
+    CService m_my_addr GUARDED_BY(m_mutex);
+
+    /**
+     * SAM session id.
+     */
+    std::string m_session_id GUARDED_BY(m_mutex);
+};
+
+} // namespace sam
+} // namespace i2p
+
+#endif /* BITCOIN_I2P_H */
diff --git a/src/init.cpp b/src/init.cpp
index befba2eb2d..f40cf7975b 100644
--- a/src/init.cpp
+++ b/src/init.cpp
@@ -447,7 +447,9 @@ void SetupServerArgs(NodeContext& node)
     argsman.AddArg("-maxtimeadjustment", strprintf("Maximum allowed median peer time offset adjustment. Local perspective of time may be influenced by peers forward or backward by this amount. (default: %u seconds)", DEFAULT_MAX_TIME_ADJUSTMENT), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-maxuploadtarget=<n>", strprintf("Tries to keep outbound traffic under the given target (in MiB per 24h). Limit does not apply to peers with 'download' permission. 0 = no limit (default: %d)", DEFAULT_MAX_UPLOAD_TARGET), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-onion=<ip:port>", "Use separate SOCKS5 proxy to reach peers via Tor onion services, set -noonion to disable (default: -proxy)", ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
-    argsman.AddArg("-onlynet=<net>", "Make outgoing connections only through network <net> (" + Join(GetNetworkNames(), ", ") + "). Incoming connections are not affected by this option. This option can be specified multiple times to allow multiple networks. Warning: if it is used with ipv4 or ipv6 but not onion and the -onion or -proxy option is set, then outbound onion connections will still be made; use -noonion or -onion=0 to disable outbound onion connections in this case.", ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
+    argsman.AddArg("-i2psam=<ip:port>", "I2P SAM proxy to reach I2P peers and accept I2P connections (default: none)", ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
+    argsman.AddArg("-i2pacceptincoming", "If set and -i2psam is also set then incoming I2P connections are accepted via the SAM proxy. If this is not set but -i2psam is set then only outgoing connections will be made to the I2P network. Ignored if -i2psam is not set. Listening for incoming I2P connections is done through the SAM proxy, not by binding to a local address and port (default: 1)", ArgsManager::ALLOW_BOOL, OptionsCategory::CONNECTION);
+    argsman.AddArg("-onlynet=<net>", "Make outgoing connections only through network <net> (" + Join(GetNetworkNames(), ", ") + "). Incoming connections are not affected by this option. This option can be specified multiple times to allow multiple networks. Warning: if it is used with non-onion networks and the -onion or -proxy option is set, then outbound onion connections will still be made; use -noonion or -onion=0 to disable outbound onion connections in this case.", ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-peerbloomfilters", strprintf("Support filtering of blocks and transaction with bloom filters (default: %u)", DEFAULT_PEERBLOOMFILTERS), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-peerblockfilters", strprintf("Serve compact block filters to peers per BIP 157 (default: %u)", DEFAULT_PEERBLOCKFILTERS), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-permitbaremultisig", strprintf("Relay non-P2SH multisig (default: %u)", DEFAULT_PERMIT_BAREMULTISIG), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
@@ -847,6 +849,9 @@ void InitParameterInteraction(ArgsManager& args)
             LogPrintf("%s: parameter interaction: -listen=0 -> setting -discover=0\n", __func__);
         if (args.SoftSetBoolArg("-listenonion", false))
             LogPrintf("%s: parameter interaction: -listen=0 -> setting -listenonion=0\n", __func__);
+        if (args.SoftSetBoolArg("-i2pacceptincoming", false)) {
+            LogPrintf("%s: parameter interaction: -listen=0 -> setting -i2pacceptincoming=0\n", __func__);
+        }
     }
 
     if (args.IsArgSet("-externalip")) {
@@ -1990,6 +1995,21 @@ bool AppInitMain(const util::Ref& context, NodeContext& node, interfaces::BlockA
             connOptions.m_specified_outgoing = connect;
         }
     }
+
+    const std::string& i2psam_arg = args.GetArg("-i2psam", "");
+    if (!i2psam_arg.empty()) {
+        CService addr;
+        if (!Lookup(i2psam_arg, addr, 7656, fNameLookup) || !addr.IsValid()) {
+            return InitError(strprintf(_("Invalid -i2psam address or hostname: '%s'"), i2psam_arg));
+        }
+        SetReachable(NET_I2P, true);
+        SetProxy(NET_I2P, proxyType{addr});
+    } else {
+        SetReachable(NET_I2P, false);
+    }
+
+    connOptions.m_i2p_accept_incoming = args.GetBoolArg("-i2pacceptincoming", true);
+
     if (!node.connman->Start(*node.scheduler, connOptions)) {
         return false;
     }
diff --git a/src/logging.cpp b/src/logging.cpp
index e82f2c2810..866213786e 100644
--- a/src/logging.cpp
+++ b/src/logging.cpp
@@ -156,6 +156,7 @@ const CLogCategoryDesc LogCategories[] =
     {BCLog::QT, "qt"},
     {BCLog::LEVELDB, "leveldb"},
     {BCLog::VALIDATION, "validation"},
+    {BCLog::I2P, "i2p"},
     {BCLog::ALL, "1"},
     {BCLog::ALL, "all"},
 };
diff --git a/src/logging.h b/src/logging.h
index 4ece8f5e3a..436f0cd12e 100644
--- a/src/logging.h
+++ b/src/logging.h
@@ -57,6 +57,7 @@ namespace BCLog {
         QT          = (1 << 19),
         LEVELDB     = (1 << 20),
         VALIDATION  = (1 << 21),
+        I2P         = (1 << 22),
         ALL         = ~(uint32_t)0,
     };
 
diff --git a/src/net.cpp b/src/net.cpp
index d9571e7036..6dcff7e32e 100644
--- a/src/net.cpp
+++ b/src/net.cpp
@@ -11,8 +11,10 @@
 
 #include <banman.h>
 #include <clientversion.h>
+#include <compat.h>
 #include <consensus/consensus.h>
 #include <crypto/sha256.h>
+#include <i2p.h>
 #include <net_permissions.h>
 #include <netbase.h>
 #include <node/ui_interface.h>
@@ -72,16 +74,6 @@ static constexpr std::chrono::seconds MAX_UPLOAD_TIMEFRAME{60 * 60 * 24};
 // We add a random period time (0 to 1 seconds) to feeler connections to prevent synchronization.
 #define FEELER_SLEEP_WINDOW 1
 
-// MSG_NOSIGNAL is not available on some platforms, if it doesn't exist define it as 0
-#if !defined(MSG_NOSIGNAL)
-#define MSG_NOSIGNAL 0
-#endif
-
-// MSG_DONTWAIT is not available on some platforms, if it doesn't exist define it as 0
-#if !defined(MSG_DONTWAIT)
-#define MSG_DONTWAIT 0
-#endif
-
 /** Used to pass flags to the Bind() function */
 enum BindFlags {
     BF_NONE         = 0,
@@ -430,10 +422,20 @@ CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCo
     bool connected = false;
     std::unique_ptr<Sock> sock;
     proxyType proxy;
+    CAddress addr_bind;
+    assert(!addr_bind.IsValid());
+
     if (addrConnect.IsValid()) {
         bool proxyConnectionFailed = false;
 
-        if (GetProxy(addrConnect.GetNetwork(), proxy)) {
+        if (addrConnect.GetNetwork() == NET_I2P && m_i2p_sam_session.get() != nullptr) {
+            i2p::Connection conn;
+            if (m_i2p_sam_session->Connect(addrConnect, conn, proxyConnectionFailed)) {
+                connected = true;
+                sock = std::make_unique<Sock>(std::move(conn.sock));
+                addr_bind = CAddress{conn.me, NODE_NONE};
+            }
+        } else if (GetProxy(addrConnect.GetNetwork(), proxy)) {
             sock = CreateSock(proxy.proxy);
             if (!sock) {
                 return nullptr;
@@ -473,7 +475,9 @@ CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCo
     // Add node
     NodeId id = GetNewNodeId();
     uint64_t nonce = GetDeterministicRandomizer(RANDOMIZER_ID_LOCALHOSTNONCE).Write(id).Finalize();
-    CAddress addr_bind = GetBindAddress(sock->Get());
+    if (!addr_bind.IsValid()) {
+        addr_bind = GetBindAddress(sock->Get());
+    }
     CNode* pnode = new CNode(id, nLocalServices, sock->Release(), addrConnect, CalculateKeyedNetGroup(addrConnect), nonce, addr_bind, pszDest ? pszDest : "", conn_type, /* inbound_onion */ false);
     pnode->AddRef();
 
@@ -1005,17 +1009,35 @@ void CConnman::AcceptConnection(const ListenSocket& hListenSocket) {
     socklen_t len = sizeof(sockaddr);
     SOCKET hSocket = accept(hListenSocket.socket, (struct sockaddr*)&sockaddr, &len);
     CAddress addr;
-    int nInbound = 0;
-    int nMaxInbound = nMaxConnections - m_max_outbound;
 
-    if (hSocket != INVALID_SOCKET) {
-        if (!addr.SetSockAddr((const struct sockaddr*)&sockaddr)) {
-            LogPrintf("Warning: Unknown socket family\n");
+    if (hSocket == INVALID_SOCKET) {
+        const int nErr = WSAGetLastError();
+        if (nErr != WSAEWOULDBLOCK) {
+            LogPrintf("socket error accept failed: %s\n", NetworkErrorString(nErr));
         }
+        return;
+    }
+
+    if (!addr.SetSockAddr((const struct sockaddr*)&sockaddr)) {
+        LogPrintf("Warning: Unknown socket family\n");
     }
 
+    const CAddress addr_bind = GetBindAddress(hSocket);
+
     NetPermissionFlags permissionFlags = NetPermissionFlags::PF_NONE;
     hListenSocket.AddSocketPermissionFlags(permissionFlags);
+
+    CreateNodeFromAcceptedSocket(hSocket, permissionFlags, addr_bind, addr);
+}
+
+void CConnman::CreateNodeFromAcceptedSocket(SOCKET hSocket,
+                                            NetPermissionFlags permissionFlags,
+                                            const CAddress& addr_bind,
+                                            const CAddress& addr)
+{
+    int nInbound = 0;
+    int nMaxInbound = nMaxConnections - m_max_outbound;
+
     AddWhitelistPermissionFlags(permissionFlags, addr);
     if (NetPermissions::HasFlag(permissionFlags, NetPermissionFlags::PF_ISIMPLICIT)) {
         NetPermissions::ClearFlag(permissionFlags, PF_ISIMPLICIT);
@@ -1032,14 +1054,6 @@ void CConnman::AcceptConnection(const ListenSocket& hListenSocket) {
         }
     }
 
-    if (hSocket == INVALID_SOCKET)
-    {
-        int nErr = WSAGetLastError();
-        if (nErr != WSAEWOULDBLOCK)
-            LogPrintf("socket error accept failed: %s\n", NetworkErrorString(nErr));
-        return;
-    }
-
     if (!fNetworkActive) {
         LogPrint(BCLog::NET, "connection from %s dropped: not accepting new connections\n", addr.ToString());
         CloseSocket(hSocket);
@@ -1087,7 +1101,6 @@ void CConnman::AcceptConnection(const ListenSocket& hListenSocket) {
 
     NodeId id = GetNewNodeId();
     uint64_t nonce = GetDeterministicRandomizer(RANDOMIZER_ID_LOCALHOSTNONCE).Write(id).Finalize();
-    CAddress addr_bind = GetBindAddress(hSocket);
 
     ServiceFlags nodeServices = nLocalServices;
     if (NetPermissions::HasFlag(permissionFlags, PF_BLOOMFILTER)) {
@@ -2175,6 +2188,45 @@ void CConnman::ThreadMessageHandler()
     }
 }
 
+void CConnman::ThreadI2PAcceptIncoming()
+{
+    static constexpr auto err_wait_begin = 1s;
+    static constexpr auto err_wait_cap = 5min;
+    auto err_wait = err_wait_begin;
+
+    bool advertising_listen_addr = false;
+    i2p::Connection conn;
+
+    while (!interruptNet) {
+
+        if (!m_i2p_sam_session->Listen(conn)) {
+            if (advertising_listen_addr && conn.me.IsValid()) {
+                RemoveLocal(conn.me);
+                advertising_listen_addr = false;
+            }
+
+            interruptNet.sleep_for(err_wait);
+            if (err_wait < err_wait_cap) {
+                err_wait *= 2;
+            }
+
+            continue;
+        }
+
+        if (!advertising_listen_addr) {
+            AddLocal(conn.me, LOCAL_BIND);
+            advertising_listen_addr = true;
+        }
+
+        if (!m_i2p_sam_session->Accept(conn)) {
+            continue;
+        }
+
+        CreateNodeFromAcceptedSocket(conn.sock.Release(), NetPermissionFlags::PF_NONE,
+                                     CAddress{conn.me, NODE_NONE}, CAddress{conn.peer, NODE_NONE});
+    }
+}
+
 bool CConnman::BindListenPort(const CService& addrBind, bilingual_str& strError, NetPermissionFlags permissions)
 {
     int nOne = 1;
@@ -2374,6 +2426,12 @@ bool CConnman::Start(CScheduler& scheduler, const Options& connOptions)
         return false;
     }
 
+    proxyType i2p_sam;
+    if (GetProxy(NET_I2P, i2p_sam)) {
+        m_i2p_sam_session = std::make_unique<i2p::sam::Session>(GetDataDir() / "i2p_private_key",
+                                                                i2p_sam.proxy, &interruptNet);
+    }
+
     for (const auto& strDest : connOptions.vSeedNodes) {
         AddAddrFetch(strDest);
     }
@@ -2454,6 +2512,12 @@ bool CConnman::Start(CScheduler& scheduler, const Options& connOptions)
     // Process messages
     threadMessageHandler = std::thread(&TraceThread<std::function<void()> >, "msghand", std::function<void()>(std::bind(&CConnman::ThreadMessageHandler, this)));
 
+    if (connOptions.m_i2p_accept_incoming && m_i2p_sam_session.get() != nullptr) {
+        threadI2PAcceptIncoming =
+            std::thread(&TraceThread<std::function<void()>>, "i2paccept",
+                        std::function<void()>(std::bind(&CConnman::ThreadI2PAcceptIncoming, this)));
+    }
+
     // Dump network addresses
     scheduler.scheduleEvery([this] { DumpAddresses(); }, DUMP_PEERS_INTERVAL);
 
@@ -2501,6 +2565,9 @@ void CConnman::Interrupt()
 
 void CConnman::StopThreads()
 {
+    if (threadI2PAcceptIncoming.joinable()) {
+        threadI2PAcceptIncoming.join();
+    }
     if (threadMessageHandler.joinable())
         threadMessageHandler.join();
     if (threadOpenConnections.joinable())
@@ -2597,9 +2664,7 @@ std::vector<CAddress> CConnman::GetAddresses(size_t max_addresses, size_t max_pc
 
 std::vector<CAddress> CConnman::GetAddresses(CNode& requestor, size_t max_addresses, size_t max_pct)
 {
-    SOCKET socket;
-    WITH_LOCK(requestor.cs_hSocket, socket = requestor.hSocket);
-    auto local_socket_bytes = GetBindAddress(socket).GetAddrBytes();
+    auto local_socket_bytes = requestor.addrBind.GetAddrBytes();
     uint64_t cache_id = GetDeterministicRandomizer(RANDOMIZER_ID_ADDRCACHE)
         .Write(requestor.addr.GetNetwork())
         .Write(local_socket_bytes.data(), local_socket_bytes.size())
diff --git a/src/net.h b/src/net.h
index 67d1cf0e55..fbdaf75150 100644
--- a/src/net.h
+++ b/src/net.h
@@ -14,6 +14,7 @@
 #include <compat.h>
 #include <crypto/siphash.h>
 #include <hash.h>
+#include <i2p.h>
 #include <net_permissions.h>
 #include <netaddress.h>
 #include <optional.h>
@@ -831,6 +832,7 @@ public:
         std::vector<std::string> m_specified_outgoing;
         std::vector<std::string> m_added_nodes;
         std::vector<bool> m_asmap;
+        bool m_i2p_accept_incoming;
     };
 
     void Init(const Options& connOptions) {
@@ -1048,7 +1050,22 @@ private:
     void ProcessAddrFetch();
     void ThreadOpenConnections(std::vector<std::string> connect);
     void ThreadMessageHandler();
+    void ThreadI2PAcceptIncoming();
     void AcceptConnection(const ListenSocket& hListenSocket);
+
+    /**
+     * Create a `CNode` object from a socket that has just been accepted and add the node to
+     * the `vNodes` member.
+     * @param[in] hSocket Connected socket to communicate with the peer.
+     * @param[in] permissionFlags The peer's permissions.
+     * @param[in] addr_bind The address and port at our side of the connection.
+     * @param[in] addr The address and port at the peer's side of the connection.
+     */
+    void CreateNodeFromAcceptedSocket(SOCKET hSocket,
+                                      NetPermissionFlags permissionFlags,
+                                      const CAddress& addr_bind,
+                                      const CAddress& addr);
+
     void DisconnectNodes();
     void NotifyNumConnectionsChanged();
     /** Return true if the peer is inactive and should be disconnected. */
@@ -1207,13 +1224,26 @@ private:
     Mutex mutexMsgProc;
     std::atomic<bool> flagInterruptMsgProc{false};
 
+    /**
+     * This is signaled when network activity should cease.
+     * A pointer to it is saved in `m_i2p_sam_session`, so make sure that
+     * the lifetime of `interruptNet` is not shorter than
+     * the lifetime of `m_i2p_sam_session`.
+     */
     CThreadInterrupt interruptNet;
 
+    /**
+     * I2P SAM session.
+     * Used to accept incoming and make outgoing I2P connections.
+     */
+    std::unique_ptr<i2p::sam::Session> m_i2p_sam_session;
+
     std::thread threadDNSAddressSeed;
     std::thread threadSocketHandler;
     std::thread threadOpenAddedConnections;
     std::thread threadOpenConnections;
     std::thread threadMessageHandler;
+    std::thread threadI2PAcceptIncoming;
 
     /** flag for deciding to connect to an extra outbound peer,
      *  in excess of m_max_outbound_full_relay
diff --git a/src/netaddress.cpp b/src/netaddress.cpp
index 85e46fd373..69edc15c66 100644
--- a/src/netaddress.cpp
+++ b/src/netaddress.cpp
@@ -221,25 +221,34 @@ static void Checksum(Span<const uint8_t> addr_pubkey, uint8_t (&checksum)[CHECKS
 
 }; // namespace torv3
 
-/**
- * Parse a TOR address and set this object to it.
- *
- * @returns Whether or not the operation was successful.
- *
- * @see CNetAddr::IsTor()
- */
-bool CNetAddr::SetSpecial(const std::string& str)
+bool CNetAddr::SetSpecial(const std::string& addr)
+{
+    if (!ValidAsCString(addr)) {
+        return false;
+    }
+
+    if (SetTor(addr)) {
+        return true;
+    }
+
+    if (SetI2P(addr)) {
+        return true;
+    }
+
+    return false;
+}
+
+bool CNetAddr::SetTor(const std::string& addr)
 {
     static const char* suffix{".onion"};
     static constexpr size_t suffix_len{6};
 
-    if (!ValidAsCString(str) || str.size() <= suffix_len ||
-        str.substr(str.size() - suffix_len) != suffix) {
+    if (addr.size() <= suffix_len || addr.substr(addr.size() - suffix_len) != suffix) {
         return false;
     }
 
     bool invalid;
-    const auto& input = DecodeBase32(str.substr(0, str.size() - suffix_len).c_str(), &invalid);
+    const auto& input = DecodeBase32(addr.substr(0, addr.size() - suffix_len).c_str(), &invalid);
 
     if (invalid) {
         return false;
@@ -275,6 +284,34 @@ bool CNetAddr::SetSpecial(const std::string& str)
     return false;
 }
 
+bool CNetAddr::SetI2P(const std::string& addr)
+{
+    // I2P addresses that we support consist of 52 base32 characters + ".b32.i2p".
+    static constexpr size_t b32_len{52};
+    static const char* suffix{".b32.i2p"};
+    static constexpr size_t suffix_len{8};
+
+    if (addr.size() != b32_len + suffix_len || ToLower(addr.substr(b32_len)) != suffix) {
+        return false;
+    }
+
+    // Remove the ".b32.i2p" suffix and pad to a multiple of 8 chars, so DecodeBase32()
+    // can decode it.
+    const std::string b32_padded = addr.substr(0, b32_len) + "====";
+
+    bool invalid;
+    const auto& address_bytes = DecodeBase32(b32_padded.c_str(), &invalid);
+
+    if (invalid || address_bytes.size() != ADDR_I2P_SIZE) {
+        return false;
+    }
+
+    m_net = NET_I2P;
+    m_addr.assign(address_bytes.begin(), address_bytes.end());
+
+    return true;
+}
+
 CNetAddr::CNetAddr(const struct in_addr& ipv4Addr)
 {
     m_net = NET_IPV4;
@@ -841,6 +878,11 @@ int CNetAddr::GetReachabilityFrom(const CNetAddr *paddrPartner) const
         case NET_IPV4:   return REACH_IPV4; // Tor users can connect to IPv4 as well
         case NET_ONION:    return REACH_PRIVATE;
         }
+    case NET_I2P:
+        switch (ourNet) {
+        case NET_I2P: return REACH_PRIVATE;
+        default: return REACH_DEFAULT;
+        }
     case NET_TEREDO:
         switch(ourNet) {
         default:          return REACH_DEFAULT;
diff --git a/src/netaddress.h b/src/netaddress.h
index d0986557f7..897ce46cda 100644
--- a/src/netaddress.h
+++ b/src/netaddress.h
@@ -151,7 +151,16 @@ class CNetAddr
 
         bool SetInternal(const std::string& name);
 
-        bool SetSpecial(const std::string &strName); // for Tor addresses
+        /**
+         * Parse a Tor or I2P address and set this object to it.
+         * @param[in] addr Address to parse, for example
+         * pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion or
+         * ukeu3k5oycgaauneqgtnvselmt4yemvoilkln7jpvamvfx7dnkdq.b32.i2p.
+         * @returns Whether the operation was successful.
+         * @see CNetAddr::IsTor(), CNetAddr::IsI2P()
+         */
+        bool SetSpecial(const std::string& addr);
+
         bool IsBindAny() const; // INADDR_ANY equivalent
         bool IsIPv4() const;    // IPv4 mapped address (::FFFF:0:0/96, 0.0.0.0/0)
         bool IsIPv6() const;    // IPv6 address (not mapped IPv4, not Tor)
@@ -249,6 +258,25 @@ class CNetAddr
 
     private:
         /**
+         * Parse a Tor address and set this object to it.
+         * @param[in] addr Address to parse, must be a valid C string, for example
+         * pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion or
+         * 6hzph5hv6337r6p2.onion.
+         * @returns Whether the operation was successful.
+         * @see CNetAddr::IsTor()
+         */
+        bool SetTor(const std::string& addr);
+
+        /**
+         * Parse an I2P address and set this object to it.
+         * @param[in] addr Address to parse, must be a valid C string, for example
+         * ukeu3k5oycgaauneqgtnvselmt4yemvoilkln7jpvamvfx7dnkdq.b32.i2p.
+         * @returns Whether the operation was successful.
+         * @see CNetAddr::IsI2P()
+         */
+        bool SetI2P(const std::string& addr);
+
+        /**
          * BIP155 network ids recognized by this software.
          */
         enum BIP155Network : uint8_t {
diff --git a/src/netbase.cpp b/src/netbase.cpp
index 0c5b3a220e..7dc616080d 100644
--- a/src/netbase.cpp
+++ b/src/netbase.cpp
@@ -5,6 +5,7 @@
 
 #include <netbase.h>
 
+#include <compat.h>
 #include <sync.h>
 #include <tinyformat.h>
 #include <util/sock.h>
@@ -14,6 +15,7 @@
 #include <util/time.h>
 
 #include <atomic>
+#include <chrono>
 #include <cstdint>
 #include <functional>
 #include <limits>
@@ -29,10 +31,6 @@
 #include <poll.h>
 #endif
 
-#if !defined(MSG_NOSIGNAL)
-#define MSG_NOSIGNAL 0
-#endif
-
 // Settings
 static Mutex g_proxyinfo_mutex;
 static proxyType proxyInfo[NET_MAX] GUARDED_BY(g_proxyinfo_mutex);
@@ -53,6 +51,9 @@ enum Network ParseNetwork(const std::string& net_in) {
         LogPrintf("Warning: net name 'tor' is deprecated and will be removed in the future. You should use 'onion' instead.\n");
         return NET_ONION;
     }
+    if (net == "i2p") {
+        return NET_I2P;
+    }
     return NET_UNROUTABLE;
 }
 
@@ -77,7 +78,7 @@ std::vector<std::string> GetNetworkNames(bool append_unroutable)
     std::vector<std::string> names;
     for (int n = 0; n < NET_MAX; ++n) {
         const enum Network network{static_cast<Network>(n)};
-        if (network == NET_UNROUTABLE || network == NET_I2P || network == NET_CJDNS || network == NET_INTERNAL) continue;
+        if (network == NET_UNROUTABLE || network == NET_CJDNS || network == NET_INTERNAL) continue;
         names.emplace_back(GetNetworkName(network));
     }
     if (append_unroutable) {
@@ -360,9 +361,6 @@ static IntrRecvError InterruptibleRecv(uint8_t* data, size_t len, int timeout, c
 {
     int64_t curTime = GetTimeMillis();
     int64_t endTime = curTime + timeout;
-    // Maximum time to wait for I/O readiness. It will take up until this time
-    // (in millis) to break off in case of an interruption.
-    const int64_t maxWait = 1000;
     while (len > 0 && curTime < endTime) {
         ssize_t ret = sock.Recv(data, len, 0); // Optimistically try the recv first
         if (ret > 0) {
@@ -373,10 +371,11 @@ static IntrRecvError InterruptibleRecv(uint8_t* data, size_t len, int timeout, c
         } else { // Other error or blocking
             int nErr = WSAGetLastError();
             if (nErr == WSAEINPROGRESS || nErr == WSAEWOULDBLOCK || nErr == WSAEINVAL) {
-                // Only wait at most maxWait milliseconds at a time, unless
+                // Only wait at most MAX_WAIT_FOR_IO at a time, unless
                 // we're approaching the end of the specified total timeout
-                int timeout_ms = std::min(endTime - curTime, maxWait);
-                if (!sock.Wait(std::chrono::milliseconds{timeout_ms}, Sock::RECV)) {
+                const auto remaining = std::chrono::milliseconds{endTime - curTime};
+                const auto timeout = std::min(remaining, std::chrono::milliseconds{MAX_WAIT_FOR_IO});
+                if (!sock.Wait(timeout, Sock::RECV)) {
                     return IntrRecvError::NetworkError;
                 }
             } else {
diff --git a/src/rpc/net.cpp b/src/rpc/net.cpp
index 0224ee697a..e80270d038 100644
--- a/src/rpc/net.cpp
+++ b/src/rpc/net.cpp
@@ -546,7 +546,7 @@ static UniValue GetNetworksInfo()
     UniValue networks(UniValue::VARR);
     for (int n = 0; n < NET_MAX; ++n) {
         enum Network network = static_cast<enum Network>(n);
-        if (network == NET_UNROUTABLE || network == NET_I2P || network == NET_CJDNS || network == NET_INTERNAL) continue;
+        if (network == NET_UNROUTABLE || network == NET_CJDNS || network == NET_INTERNAL) continue;
         proxyType proxy;
         UniValue obj(UniValue::VOBJ);
         GetProxy(network, proxy);
diff --git a/src/test/net_tests.cpp b/src/test/net_tests.cpp
index 1c7c35528e..f039f76f82 100644
--- a/src/test/net_tests.cpp
+++ b/src/test/net_tests.cpp
@@ -322,6 +322,7 @@ BOOST_AUTO_TEST_CASE(cnetaddr_basic)
     BOOST_REQUIRE(addr.IsValid());
     BOOST_REQUIRE(addr.IsTor());
 
+    BOOST_CHECK(!addr.IsI2P());
     BOOST_CHECK(!addr.IsBindAny());
     BOOST_CHECK(addr.IsAddrV1Compatible());
     BOOST_CHECK_EQUAL(addr.ToString(), "6hzph5hv6337r6p2.onion");
@@ -332,6 +333,7 @@ BOOST_AUTO_TEST_CASE(cnetaddr_basic)
     BOOST_REQUIRE(addr.IsValid());
     BOOST_REQUIRE(addr.IsTor());
 
+    BOOST_CHECK(!addr.IsI2P());
     BOOST_CHECK(!addr.IsBindAny());
     BOOST_CHECK(!addr.IsAddrV1Compatible());
     BOOST_CHECK_EQUAL(addr.ToString(), torv3_addr);
@@ -352,6 +354,35 @@ BOOST_AUTO_TEST_CASE(cnetaddr_basic)
     // TOR, invalid base32
     BOOST_CHECK(!addr.SetSpecial(std::string{"mf*g zak.onion"}));
 
+    // I2P
+    const char* i2p_addr = "UDHDrtrcetjm5sxzskjyr5ztpeszydbh4dpl3pl4utgqqw2v4jna.b32.I2P";
+    BOOST_REQUIRE(addr.SetSpecial(i2p_addr));
+    BOOST_REQUIRE(addr.IsValid());
+    BOOST_REQUIRE(addr.IsI2P());
+
+    BOOST_CHECK(!addr.IsTor());
+    BOOST_CHECK(!addr.IsBindAny());
+    BOOST_CHECK(!addr.IsAddrV1Compatible());
+    BOOST_CHECK_EQUAL(addr.ToString(), ToLower(i2p_addr));
+
+    // I2P, correct length, but decodes to less than the expected number of bytes.
+    BOOST_CHECK(!addr.SetSpecial("udhdrtrcetjm5sxzskjyr5ztpeszydbh4dpl3pl4utgqqw2v4jn=.b32.i2p"));
+
+    // I2P, extra unnecessary padding
+    BOOST_CHECK(!addr.SetSpecial("udhdrtrcetjm5sxzskjyr5ztpeszydbh4dpl3pl4utgqqw2v4jna=.b32.i2p"));
+
+    // I2P, malicious
+    BOOST_CHECK(!addr.SetSpecial("udhdrtrcetjm5sxzskjyr5ztpeszydbh4dpl3pl4utgqqw2v\0wtf.b32.i2p"s));
+
+    // I2P, valid but unsupported (56 Base32 characters)
+    // See "Encrypted LS with Base 32 Addresses" in
+    // https://geti2p.net/spec/encryptedleaseset.txt
+    BOOST_CHECK(
+        !addr.SetSpecial("pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscsad.b32.i2p"));
+
+    // I2P, invalid base32
+    BOOST_CHECK(!addr.SetSpecial(std::string{"tp*szydbh4dp.b32.i2p"}));
+
     // Internal
     addr.SetInternal("esffpp");
     BOOST_REQUIRE(!addr.IsValid()); // "internal" is considered invalid
diff --git a/src/torcontrol.cpp b/src/torcontrol.cpp
index 605c77fc3a..0f2a4895f1 100644
--- a/src/torcontrol.cpp
+++ b/src/torcontrol.cpp
@@ -12,6 +12,7 @@
 #include <net.h>
 #include <netaddress.h>
 #include <netbase.h>
+#include <util/readwritefile.h>
 #include <util/strencodings.h>
 #include <util/system.h>
 #include <util/time.h>
@@ -362,52 +363,6 @@ std::map<std::string,std::string> ParseTorReplyMapping(const std::string &s)
     return mapping;
 }
 
-/** Read full contents of a file and return them in a std::string.
- * Returns a pair <status, string>.
- * If an error occurred, status will be false, otherwise status will be true and the data will be returned in string.
- *
- * @param maxsize Puts a maximum size limit on the file that is read. If the file is larger than this, truncated data
- *         (with len > maxsize) will be returned.
- */
-static std::pair<bool,std::string> ReadBinaryFile(const fs::path &filename, size_t maxsize=std::numeric_limits<size_t>::max())
-{
-    FILE *f = fsbridge::fopen(filename, "rb");
-    if (f == nullptr)
-        return std::make_pair(false,"");
-    std::string retval;
-    char buffer[128];
-    size_t n;
-    while ((n=fread(buffer, 1, sizeof(buffer), f)) > 0) {
-        // Check for reading errors so we don't return any data if we couldn't
-        // read the entire file (or up to maxsize)
-        if (ferror(f)) {
-            fclose(f);
-            return std::make_pair(false,"");
-        }
-        retval.append(buffer, buffer+n);
-        if (retval.size() > maxsize)
-            break;
-    }
-    fclose(f);
-    return std::make_pair(true,retval);
-}
-
-/** Write contents of std::string to a file.
- * @return true on success.
- */
-static bool WriteBinaryFile(const fs::path &filename, const std::string &data)
-{
-    FILE *f = fsbridge::fopen(filename, "wb");
-    if (f == nullptr)
-        return false;
-    if (fwrite(data.data(), 1, data.size(), f) != data.size()) {
-        fclose(f);
-        return false;
-    }
-    fclose(f);
-    return true;
-}
-
 /****** Bitcoin specific TorController implementation ********/
 
 /** Controller that connects to Tor control socket, authenticate, then create
diff --git a/src/util/readwritefile.cpp b/src/util/readwritefile.cpp
new file mode 100644
index 0000000000..a45c41d367
--- /dev/null
+++ b/src/util/readwritefile.cpp
@@ -0,0 +1,47 @@
+// Copyright (c) 2015-2020 The Bitcoin Core developers
+// Copyright (c) 2017 The Zcash developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <fs.h>
+
+#include <limits>
+#include <stdio.h>
+#include <string>
+#include <utility>
+
+std::pair<bool,std::string> ReadBinaryFile(const fs::path &filename, size_t maxsize=std::numeric_limits<size_t>::max())
+{
+    FILE *f = fsbridge::fopen(filename, "rb");
+    if (f == nullptr)
+        return std::make_pair(false,"");
+    std::string retval;
+    char buffer[128];
+    do {
+        const size_t n = fread(buffer, 1, sizeof(buffer), f);
+        // Check for reading errors so we don't return any data if we couldn't
+        // read the entire file (or up to maxsize)
+        if (ferror(f)) {
+            fclose(f);
+            return std::make_pair(false,"");
+        }
+        retval.append(buffer, buffer+n);
+    } while (!feof(f) && retval.size() <= maxsize);
+    fclose(f);
+    return std::make_pair(true,retval);
+}
+
+bool WriteBinaryFile(const fs::path &filename, const std::string &data)
+{
+    FILE *f = fsbridge::fopen(filename, "wb");
+    if (f == nullptr)
+        return false;
+    if (fwrite(data.data(), 1, data.size(), f) != data.size()) {
+        fclose(f);
+        return false;
+    }
+    if (fclose(f) != 0) {
+        return false;
+    }
+    return true;
+}
diff --git a/src/util/readwritefile.h b/src/util/readwritefile.h
new file mode 100644
index 0000000000..1dab874b38
--- /dev/null
+++ b/src/util/readwritefile.h
@@ -0,0 +1,28 @@
+// Copyright (c) 2015-2020 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_UTIL_READWRITEFILE_H
+#define BITCOIN_UTIL_READWRITEFILE_H
+
+#include <fs.h>
+
+#include <limits>
+#include <string>
+#include <utility>
+
+/** Read full contents of a file and return them in a std::string.
+ * Returns a pair <status, string>.
+ * If an error occurred, status will be false, otherwise status will be true and the data will be returned in string.
+ *
+ * @param maxsize Puts a maximum size limit on the file that is read. If the file is larger than this, truncated data
+ *         (with len > maxsize) will be returned.
+ */
+std::pair<bool,std::string> ReadBinaryFile(const fs::path &filename, size_t maxsize=std::numeric_limits<size_t>::max());
+
+/** Write contents of std::string to a file.
+ * @return true on success.
+ */
+bool WriteBinaryFile(const fs::path &filename, const std::string &data);
+
+#endif /* BITCOIN_UTIL_READWRITEFILE_H */
diff --git a/src/util/sock.cpp b/src/util/sock.cpp
index 4c65b5b680..e13c52a16a 100644
--- a/src/util/sock.cpp
+++ b/src/util/sock.cpp
@@ -4,6 +4,7 @@
 
 #include <compat.h>
 #include <logging.h>
+#include <threadinterrupt.h>
 #include <tinyformat.h>
 #include <util/sock.h>
 #include <util/system.h>
@@ -12,12 +13,18 @@
 #include <codecvt>
 #include <cwchar>
 #include <locale>
+#include <stdexcept>
 #include <string>
 
 #ifdef USE_POLL
 #include <poll.h>
 #endif
 
+static inline bool IOErrorIsPermanent(int err)
+{
+    return err != WSAEAGAIN && err != WSAEINTR && err != WSAEWOULDBLOCK && err != WSAEINPROGRESS;
+}
+
 Sock::Sock() : m_socket(INVALID_SOCKET) {}
 
 Sock::Sock(SOCKET s) : m_socket(s) {}
@@ -59,7 +66,7 @@ ssize_t Sock::Recv(void* buf, size_t len, int flags) const
     return recv(m_socket, static_cast<char*>(buf), len, flags);
 }
 
-bool Sock::Wait(std::chrono::milliseconds timeout, Event requested) const
+bool Sock::Wait(std::chrono::milliseconds timeout, Event requested, Event* occurred) const
 {
 #ifdef USE_POLL
     pollfd fd;
@@ -72,7 +79,21 @@ bool Sock::Wait(std::chrono::milliseconds timeout, Event requested) const
         fd.events |= POLLOUT;
     }
 
-    return poll(&fd, 1, count_milliseconds(timeout)) != SOCKET_ERROR;
+    if (poll(&fd, 1, count_milliseconds(timeout)) == SOCKET_ERROR) {
+        return false;
+    }
+
+    if (occurred != nullptr) {
+        *occurred = 0;
+        if (fd.revents & POLLIN) {
+            *occurred |= RECV;
+        }
+        if (fd.revents & POLLOUT) {
+            *occurred |= SEND;
+        }
+    }
+
+    return true;
 #else
     if (!IsSelectableSocket(m_socket)) {
         return false;
@@ -93,10 +114,167 @@ bool Sock::Wait(std::chrono::milliseconds timeout, Event requested) const
 
     timeval timeout_struct = MillisToTimeval(timeout);
 
-    return select(m_socket + 1, &fdset_recv, &fdset_send, nullptr, &timeout_struct) != SOCKET_ERROR;
+    if (select(m_socket + 1, &fdset_recv, &fdset_send, nullptr, &timeout_struct) == SOCKET_ERROR) {
+        return false;
+    }
+
+    if (occurred != nullptr) {
+        *occurred = 0;
+        if (FD_ISSET(m_socket, &fdset_recv)) {
+            *occurred |= RECV;
+        }
+        if (FD_ISSET(m_socket, &fdset_send)) {
+            *occurred |= SEND;
+        }
+    }
+
+    return true;
 #endif /* USE_POLL */
 }
 
+void Sock::SendComplete(const std::string& data,
+                        std::chrono::milliseconds timeout,
+                        CThreadInterrupt& interrupt) const
+{
+    const auto deadline = GetTime<std::chrono::milliseconds>() + timeout;
+    size_t sent{0};
+
+    for (;;) {
+        const ssize_t ret{Send(data.data() + sent, data.size() - sent, MSG_NOSIGNAL)};
+
+        if (ret > 0) {
+            sent += static_cast<size_t>(ret);
+            if (sent == data.size()) {
+                break;
+            }
+        } else {
+            const int err{WSAGetLastError()};
+            if (IOErrorIsPermanent(err)) {
+                throw std::runtime_error(strprintf("send(): %s", NetworkErrorString(err)));
+            }
+        }
+
+        const auto now = GetTime<std::chrono::milliseconds>();
+
+        if (now >= deadline) {
+            throw std::runtime_error(strprintf(
+                "Send timeout (sent only %u of %u bytes before that)", sent, data.size()));
+        }
+
+        if (interrupt) {
+            throw std::runtime_error(strprintf(
+                "Send interrupted (sent only %u of %u bytes before that)", sent, data.size()));
+        }
+
+        // Wait for a short while (or the socket to become ready for sending) before retrying
+        // if nothing was sent.
+        const auto wait_time = std::min(deadline - now, std::chrono::milliseconds{MAX_WAIT_FOR_IO});
+        Wait(wait_time, SEND);
+    }
+}
+
+std::string Sock::RecvUntilTerminator(uint8_t terminator,
+                                      std::chrono::milliseconds timeout,
+                                      CThreadInterrupt& interrupt) const
+{
+    const auto deadline = GetTime<std::chrono::milliseconds>() + timeout;
+    std::string data;
+    bool terminator_found{false};
+
+    // We must not consume any bytes past the terminator from the socket.
+    // One option is to read one byte at a time and check if we have read a terminator.
+    // However that is very slow. Instead, we peek at what is in the socket and only read
+    // as many bytes as possible without crossing the terminator.
+    // Reading 64 MiB of random data with 262526 terminator chars takes 37 seconds to read
+    // one byte at a time VS 0.71 seconds with the "peek" solution below. Reading one byte
+    // at a time is about 50 times slower.
+
+    for (;;) {
+        char buf[512];
+
+        const ssize_t peek_ret{Recv(buf, sizeof(buf), MSG_PEEK)};
+
+        switch (peek_ret) {
+        case -1: {
+            const int err{WSAGetLastError()};
+            if (IOErrorIsPermanent(err)) {
+                throw std::runtime_error(strprintf("recv(): %s", NetworkErrorString(err)));
+            }
+            break;
+        }
+        case 0:
+            throw std::runtime_error("Connection unexpectedly closed by peer");
+        default:
+            auto end = buf + peek_ret;
+            auto terminator_pos = std::find(buf, end, terminator);
+            terminator_found = terminator_pos != end;
+
+            const size_t try_len{terminator_found ? terminator_pos - buf + 1 :
+                                                    static_cast<size_t>(peek_ret)};
+
+            const ssize_t read_ret{Recv(buf, try_len, 0)};
+
+            if (read_ret < 0 || static_cast<size_t>(read_ret) != try_len) {
+                throw std::runtime_error(
+                    strprintf("recv() returned %u bytes on attempt to read %u bytes but previous "
+                              "peek claimed %u bytes are available",
+                              read_ret, try_len, peek_ret));
+            }
+
+            // Don't include the terminator in the output.
+            const size_t append_len{terminator_found ? try_len - 1 : try_len};
+
+            data.append(buf, buf + append_len);
+
+            if (terminator_found) {
+                return data;
+            }
+        }
+
+        const auto now = GetTime<std::chrono::milliseconds>();
+
+        if (now >= deadline) {
+            throw std::runtime_error(strprintf(
+                "Receive timeout (received %u bytes without terminator before that)", data.size()));
+        }
+
+        if (interrupt) {
+            throw std::runtime_error(strprintf(
+                "Receive interrupted (received %u bytes without terminator before that)",
+                data.size()));
+        }
+
+        // Wait for a short while (or the socket to become ready for reading) before retrying.
+        const auto wait_time = std::min(deadline - now, std::chrono::milliseconds{MAX_WAIT_FOR_IO});
+        Wait(wait_time, RECV);
+    }
+}
+
+bool Sock::IsConnected(std::string& errmsg) const
+{
+    if (m_socket == INVALID_SOCKET) {
+        errmsg = "not connected";
+        return false;
+    }
+
+    char c;
+    switch (Recv(&c, sizeof(c), MSG_PEEK)) {
+    case -1: {
+        const int err = WSAGetLastError();
+        if (IOErrorIsPermanent(err)) {
+            errmsg = NetworkErrorString(err);
+            return false;
+        }
+        return true;
+    }
+    case 0:
+        errmsg = "closed";
+        return false;
+    default:
+        return true;
+    }
+}
+
 #ifdef WIN32
 std::string NetworkErrorString(int err)
 {
diff --git a/src/util/sock.h b/src/util/sock.h
index 26fe60f18f..ecebb84205 100644
--- a/src/util/sock.h
+++ b/src/util/sock.h
@@ -6,11 +6,19 @@
 #define BITCOIN_UTIL_SOCK_H
 
 #include <compat.h>
+#include <threadinterrupt.h>
+#include <util/time.h>
 
 #include <chrono>
 #include <string>
 
 /**
+ * Maximum time to wait for I/O readiness.
+ * It will take up until this time to break off in case of an interruption.
+ */
+static constexpr auto MAX_WAIT_FOR_IO = 1s;
+
+/**
  * RAII helper class that manages a socket. Mimics `std::unique_ptr`, but instead of a pointer it
  * contains a socket and closes it automatically when it goes out of scope.
  */
@@ -98,9 +106,49 @@ public:
      * Wait for readiness for input (recv) or output (send).
      * @param[in] timeout Wait this much for at least one of the requested events to occur.
      * @param[in] requested Wait for those events, bitwise-or of `RECV` and `SEND`.
+     * @param[out] occurred If not nullptr and `true` is returned, then upon return this
+     * indicates which of the requested events occurred. A timeout is indicated by return
+     * value of `true` and `occurred` being set to 0.
      * @return true on success and false otherwise
      */
-    virtual bool Wait(std::chrono::milliseconds timeout, Event requested) const;
+    virtual bool Wait(std::chrono::milliseconds timeout,
+                      Event requested,
+                      Event* occurred = nullptr) const;
+
+    /* Higher level, convenience, methods. These may throw. */
+
+    /**
+     * Send the given data, retrying on transient errors.
+     * @param[in] data Data to send.
+     * @param[in] timeout Timeout for the entire operation.
+     * @param[in] interrupt If this is signaled then the operation is canceled.
+     * @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,
+                              std::chrono::milliseconds timeout,
+                              CThreadInterrupt& interrupt) const;
+
+    /**
+     * Read from socket until a terminator character is encountered. Will never consume bytes past
+     * the terminator from the socket.
+     * @param[in] terminator Character up to which to read from the socket.
+     * @param[in] timeout Timeout for the entire operation.
+     * @param[in] interrupt If this is signaled then the operation is canceled.
+     * @return The data that has been read, without the terminating character.
+     * @throws std::runtime_error if the operation cannot be completed. In this case some bytes may
+     * have been consumed from the socket.
+     */
+    virtual std::string RecvUntilTerminator(uint8_t terminator,
+                                            std::chrono::milliseconds timeout,
+                                            CThreadInterrupt& interrupt) const;
+
+    /**
+     * Check if still connected.
+     * @param[out] err The error string, if the socket has been disconnected.
+     * @return true if connected
+     */
+    virtual bool IsConnected(std::string& errmsg) const;
 
 private:
     /**