Merge #6276: backport: merge bitcoin#22817, #23042, #22777, #23774, #25443, #26138, #26854, #27128, #27761, #27863, #28287, #30118, partial bitcoin#22778 (auxiliary backports: part 16)

e458adb61c merge bitcoin#30118: improve robustness of connect_nodes() (UdjinM6)
ac94de23ae merge bitcoin#28287: add `sendmsgtopeer` rpc and a test for net-level deadlock situation (Kittywhiskers Van Gogh)
d1fce0b7ca fix: ensure that deadlocks are actually resolved (Kittywhiskers Van Gogh)
19e7bf64c8 merge bitcoin#27863: do not break when addr is not from a distinct network group (Kittywhiskers Van Gogh)
1adb9a232c merge bitcoin#27761: Log addresses of stalling peers (Kittywhiskers Van Gogh)
2854a6aa5a merge bitcoin#27128: fix intermittent issue in `p2p_disconnect_ban` (Kittywhiskers Van Gogh)
d4b0faeae1 merge bitcoin#26854: Fix intermittent timeout in p2p_permissions.py (Kittywhiskers Van Gogh)
892e329ada merge bitcoin#26138: Avoid race in disconnect_nodes helper (Kittywhiskers Van Gogh)
d6ce037814 merge bitcoin#25443: Fail if connect_nodes fails (Kittywhiskers Van Gogh)
60b5392d92 partial bitcoin#22778: Reduce resource usage for inbound block-relay-only connections (Kittywhiskers Van Gogh)
85c4aef9cb merge bitcoin#23774: Add missing assert_equal import to p2p_add_connections.py (Kittywhiskers Van Gogh)
03544175d9 merge bitcoin#22777: don't request tx relay on feeler connections (Kittywhiskers Van Gogh)
7229eb0ae2 merge bitcoin#23042: Avoid logging AlreadyHaveTx when disconnecting misbehaving peer (Kittywhiskers Van Gogh)
05395ff37b merge bitcoin#22817: Avoid race after connect_nodes (Kittywhiskers Van Gogh)

Pull request description:

  ## Additional Information

  * Depends on https://github.com/dashpay/dash/pull/6286

  * Depends on https://github.com/dashpay/dash/pull/6287

  * Depends on https://github.com/dashpay/dash/pull/6289

  * When backporting [bitcoin#28287](https://github.com/bitcoin/bitcoin/pull/28287), `p2p_net_deadlock.py` relies on the function, `random_bytes()`, that is introduced in [bitcoin#25625](https://github.com/bitcoin/bitcoin/pull/25625). Backporting [bitcoin#25625](https://github.com/bitcoin/bitcoin/pull/25625) would attract changes outside the scope of this PR.

    In the interest of brevity, the changes that introduce `random_bytes()` have been included in [bitcoin#28287](https://github.com/bitcoin/bitcoin/pull/28287) instead.

  ## Breaking Changes

  None expected.

  ## Checklist:

  - [x] I have performed a self-review of my own code
  - [x] I have commented my code, particularly in hard-to-understand areas **(note: N/A)**
  - [x] I have added or updated relevant unit/integration/functional/e2e tests
  - [x] I have made corresponding changes to the documentation **(note: N/A)**
  - [x] I have assigned this pull request to a milestone _(for repository code-owners and collaborators only)_

ACKs for top commit:
  UdjinM6:
    utACK e458adb61c
  PastaPastaPasta:
    utACK e458adb61c

Tree-SHA512: 48494004dddecb31c53f5e19ab0114b92ed7b4381c7977800fd49b7403222badbfdcfe46241e854f5b086c6f54a35f6483f91c6f047b7ac9b1e88e35bb32ad02
This commit is contained in:
pasta 2024-10-04 14:08:12 -05:00
commit 8cef87d81d
No known key found for this signature in database
GPG Key ID: E2F3D7916E722D38
14 changed files with 242 additions and 56 deletions

View File

@ -1998,7 +1998,6 @@ bool CConnman::AddConnection(const std::string& address, ConnectionType conn_typ
switch (conn_type) { switch (conn_type) {
case ConnectionType::INBOUND: case ConnectionType::INBOUND:
case ConnectionType::MANUAL: case ConnectionType::MANUAL:
case ConnectionType::FEELER:
return false; return false;
case ConnectionType::OUTBOUND_FULL_RELAY: case ConnectionType::OUTBOUND_FULL_RELAY:
max_connections = m_max_outbound_full_relay; max_connections = m_max_outbound_full_relay;
@ -2009,6 +2008,9 @@ bool CConnman::AddConnection(const std::string& address, ConnectionType conn_typ
// no limit for ADDR_FETCH because -seednode has no limit either // no limit for ADDR_FETCH because -seednode has no limit either
case ConnectionType::ADDR_FETCH: case ConnectionType::ADDR_FETCH:
break; break;
// no limit for FEELER connections since they're short-lived
case ConnectionType::FEELER:
break;
} // no default case, so the compiler can warn about missing cases } // no default case, so the compiler can warn about missing cases
// Count existing connections // Count existing connections
@ -2253,6 +2255,7 @@ bool CConnman::GenerateSelectSet(const std::vector<CNode*>& nodes,
for (CNode* pnode : nodes) { for (CNode* pnode : nodes) {
bool select_recv = !pnode->fHasRecvData; bool select_recv = !pnode->fHasRecvData;
bool select_send = !pnode->fCanSendData; bool select_send = !pnode->fCanSendData;
if (!select_recv && !select_send) continue;
LOCK(pnode->m_sock_mutex); LOCK(pnode->m_sock_mutex);
if (!pnode->m_sock) { if (!pnode->m_sock) {
@ -2623,9 +2626,7 @@ void CConnman::SocketHandlerConnected(const std::set<SOCKET>& recv_set,
// receiving data. This means properly utilizing TCP flow control signalling. // receiving data. This means properly utilizing TCP flow control signalling.
// * Otherwise, if there is space left in the receive buffer (!fPauseRecv), try // * Otherwise, if there is space left in the receive buffer (!fPauseRecv), try
// receiving data (which should succeed as the socket signalled as receivable). // receiving data (which should succeed as the socket signalled as receivable).
const auto& [to_send, more, _msg_type] = it->second->m_transport->GetBytesToSend(it->second->nSendMsgSize != 0); if (!it->second->fPauseRecv && !it->second->fDisconnect && it->second->nSendMsgSize == 0) {
const bool queue_is_empty{to_send.empty() && !more};
if (!it->second->fPauseRecv && !it->second->fDisconnect && queue_is_empty) {
it->second->AddRef(); it->second->AddRef();
vReceivableNodes.emplace(it->second); vReceivableNodes.emplace(it->second);
} }
@ -3301,7 +3302,7 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, CDe
// Require outbound IPv4/IPv6 connections, other than feelers, to be to distinct network groups // Require outbound IPv4/IPv6 connections, other than feelers, to be to distinct network groups
if (!fFeeler && outbound_ipv46_peer_netgroups.count(m_netgroupman.GetGroup(addr))) { if (!fFeeler && outbound_ipv46_peer_netgroups.count(m_netgroupman.GetGroup(addr))) {
break; continue;
} }
// if we selected an invalid address, restart // if we selected an invalid address, restart

View File

@ -1433,8 +1433,8 @@ public:
* Attempts to open a connection. Currently only used from tests. * Attempts to open a connection. Currently only used from tests.
* *
* @param[in] address Address of node to try connecting to * @param[in] address Address of node to try connecting to
* @param[in] conn_type ConnectionType::OUTBOUND or ConnectionType::BLOCK_RELAY * @param[in] conn_type ConnectionType::OUTBOUND, ConnectionType::BLOCK_RELAY,
* or ConnectionType::ADDR_FETCH * ConnectionType::ADDR_FETCH or ConnectionType::FEELER
* @return bool Returns false if there are no available * @return bool Returns false if there are no available
* slots for this connection: * slots for this connection:
* - conn_type not a supported ConnectionType * - conn_type not a supported ConnectionType

View File

@ -272,25 +272,34 @@ struct Peer {
struct TxRelay { struct TxRelay {
mutable RecursiveMutex m_bloom_filter_mutex; mutable RecursiveMutex m_bloom_filter_mutex;
// We use m_relay_txs for two purposes - /** Whether the peer wishes to receive transaction announcements.
// a) it allows us to not relay tx invs before receiving the peer's version message *
// b) the peer may tell us in its version message that we should not relay tx invs * This is initially set based on the fRelay flag in the received
// unless it loads a bloom filter. * `version` message. If initially set to false, it can only be flipped
* to true if we have offered the peer NODE_BLOOM services and it sends
* us a `filterload` or `filterclear` message. See BIP37. */
bool m_relay_txs GUARDED_BY(m_bloom_filter_mutex){false}; bool m_relay_txs GUARDED_BY(m_bloom_filter_mutex){false};
/** A bloom filter for which transactions to announce to the peer. See BIP37. */
std::unique_ptr<CBloomFilter> m_bloom_filter PT_GUARDED_BY(m_bloom_filter_mutex) GUARDED_BY(m_bloom_filter_mutex){nullptr}; std::unique_ptr<CBloomFilter> m_bloom_filter PT_GUARDED_BY(m_bloom_filter_mutex) GUARDED_BY(m_bloom_filter_mutex){nullptr};
mutable RecursiveMutex m_tx_inventory_mutex; mutable RecursiveMutex m_tx_inventory_mutex;
// inventory based relay /** A filter of all the txids that the peer has announced to
* us or we have announced to the peer. We use this to avoid announcing
* the same txid to a peer that already has the transaction. */
CRollingBloomFilter m_tx_inventory_known_filter GUARDED_BY(m_tx_inventory_mutex){50000, 0.000001}; CRollingBloomFilter m_tx_inventory_known_filter GUARDED_BY(m_tx_inventory_mutex){50000, 0.000001};
// Set of transaction ids we still have to announce. /** Set of transaction ids we still have to announce. We use the
// They are sorted by the mempool before relay, so the order is not important. * mempool to sort transactions in dependency order before relay, so
* this does not have to be sorted. */
std::set<uint256> m_tx_inventory_to_send GUARDED_BY(m_tx_inventory_mutex); std::set<uint256> m_tx_inventory_to_send GUARDED_BY(m_tx_inventory_mutex);
// List of non-tx/non-block inventory items /** List of non-tx/non-block inventory items */
std::vector<CInv> vInventoryOtherToSend GUARDED_BY(m_tx_inventory_mutex); std::vector<CInv> vInventoryOtherToSend GUARDED_BY(m_tx_inventory_mutex);
// Used for BIP35 mempool sending, also protected by m_tx_inventory_mutex /** Whether the peer has requested us to send our complete mempool. Only
* permitted if the peer has NetPermissionFlags::Mempool. See BIP35. */
bool m_send_mempool GUARDED_BY(m_tx_inventory_mutex){false}; bool m_send_mempool GUARDED_BY(m_tx_inventory_mutex){false};
// Last time a "MEMPOOL" request was serviced. /** The last time a BIP35 `mempool` request was serviced. */
std::atomic<std::chrono::seconds> m_last_mempool_req{0s}; std::atomic<std::chrono::seconds> m_last_mempool_req{0s};
/** The next time after which we will send an `inv` message containing
* transaction announcements to this peer. */
std::chrono::microseconds m_next_inv_send_time GUARDED_BY(NetEventsInterface::g_msgproc_mutex){0}; std::chrono::microseconds m_next_inv_send_time GUARDED_BY(NetEventsInterface::g_msgproc_mutex){0};
}; };
@ -1384,7 +1393,7 @@ void PeerManagerImpl::PushNodeVersion(CNode& pnode, const Peer& peer)
nProtocolVersion = gArgs.GetArg("-pushversion", PROTOCOL_VERSION); nProtocolVersion = gArgs.GetArg("-pushversion", PROTOCOL_VERSION);
} }
const bool tx_relay = !m_ignore_incoming_txs && !pnode.IsBlockOnlyConn(); const bool tx_relay = !m_ignore_incoming_txs && !pnode.IsBlockOnlyConn() && !pnode.IsFeelerConn();
m_connman.PushMessage(&pnode, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::VERSION, nProtocolVersion, my_services, nTime, m_connman.PushMessage(&pnode, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::VERSION, nProtocolVersion, my_services, nTime,
your_services, addr_you, // Together the pre-version-31402 serialization of CAddress "addrYou" (without nTime) your_services, addr_you, // Together the pre-version-31402 serialization of CAddress "addrYou" (without nTime)
my_services, CService(), // Together the pre-version-31402 serialization of CAddress "addrMe" (without nTime) my_services, CService(), // Together the pre-version-31402 serialization of CAddress "addrMe" (without nTime)
@ -3831,6 +3840,12 @@ void PeerManagerImpl::ProcessMessage(
best_block = &inv.hash; best_block = &inv.hash;
} }
} else { } else {
if (fBlocksOnly && NetMessageViolatesBlocksOnly(inv.GetCommand())) {
LogPrint(BCLog::NET, "%s (%s) inv sent in violation of protocol, disconnecting peer=%d\n", inv.GetCommand(), inv.hash.ToString(), pfrom.GetId());
pfrom.fDisconnect = true;
return;
}
const bool fAlreadyHave = AlreadyHave(inv); const bool fAlreadyHave = AlreadyHave(inv);
LogPrint(BCLog::NET, "got inv: %s %s peer=%d\n", inv.ToString(), fAlreadyHave ? "have" : "new", pfrom.GetId()); LogPrint(BCLog::NET, "got inv: %s %s peer=%d\n", inv.ToString(), fAlreadyHave ? "have" : "new", pfrom.GetId());
::g_stats_client->inc(strprintf("message.received.inv_%s", inv.GetCommand()), 1.0f); ::g_stats_client->inc(strprintf("message.received.inv_%s", inv.GetCommand()), 1.0f);
@ -3840,11 +3855,7 @@ void PeerManagerImpl::ProcessMessage(
}; };
AddKnownInv(*peer, inv.hash); AddKnownInv(*peer, inv.hash);
if (fBlocksOnly && NetMessageViolatesBlocksOnly(inv.GetCommand())) { if (!fAlreadyHave) {
LogPrint(BCLog::NET, "%s (%s) inv sent in violation of protocol, disconnecting peer=%d\n", inv.GetCommand(), inv.hash.ToString(), pfrom.GetId());
pfrom.fDisconnect = true;
return;
} else if (!fAlreadyHave) {
if (fBlocksOnly && inv.type == MSG_ISDLOCK) { if (fBlocksOnly && inv.type == MSG_ISDLOCK) {
if (pfrom.GetCommonVersion() <= ADDRV2_PROTO_VERSION) { if (pfrom.GetCommonVersion() <= ADDRV2_PROTO_VERSION) {
// It's ok to receive these invs, we just ignore them // It's ok to receive these invs, we just ignore them
@ -5889,7 +5900,7 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
// Stalling only triggers when the block download window cannot move. During normal steady state, // Stalling only triggers when the block download window cannot move. During normal steady state,
// the download window should be much larger than the to-be-downloaded set of blocks, so disconnection // the download window should be much larger than the to-be-downloaded set of blocks, so disconnection
// should only happen during initial block download. // should only happen during initial block download.
LogPrintf("Peer=%d is stalling block download, disconnecting\n", pto->GetId()); LogPrintf("Peer=%d%s is stalling block download, disconnecting\n", pto->GetId(), fLogIPs ? strprintf(" peeraddr=%s", pto->addr.ToStringAddrPort()) : "");
pto->fDisconnect = true; pto->fDisconnect = true;
// Increase timeout for the next peer so that we don't disconnect multiple peers if our own // Increase timeout for the next peer so that we don't disconnect multiple peers if our own
// bandwidth is insufficient. // bandwidth is insufficient.
@ -5908,7 +5919,7 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
QueuedBlock &queuedBlock = state.vBlocksInFlight.front(); QueuedBlock &queuedBlock = state.vBlocksInFlight.front();
int nOtherPeersWithValidatedDownloads = m_peers_downloading_from - 1; int nOtherPeersWithValidatedDownloads = m_peers_downloading_from - 1;
if (current_time > state.m_downloading_since + std::chrono::seconds{consensusParams.nPowTargetSpacing} * (BLOCK_DOWNLOAD_TIMEOUT_BASE + BLOCK_DOWNLOAD_TIMEOUT_PER_PEER * nOtherPeersWithValidatedDownloads)) { if (current_time > state.m_downloading_since + std::chrono::seconds{consensusParams.nPowTargetSpacing} * (BLOCK_DOWNLOAD_TIMEOUT_BASE + BLOCK_DOWNLOAD_TIMEOUT_PER_PEER * nOtherPeersWithValidatedDownloads)) {
LogPrintf("Timeout downloading block %s from peer=%d, disconnecting\n", queuedBlock.pindex->GetBlockHash().ToString(), pto->GetId()); LogPrintf("Timeout downloading block %s from peer=%d%s, disconnecting\n", queuedBlock.pindex->GetBlockHash().ToString(), pto->GetId(), fLogIPs ? strprintf(" peeraddr=%s", pto->addr.ToStringAddrPort()) : "");
pto->fDisconnect = true; pto->fDisconnect = true;
return true; return true;
} }
@ -5924,11 +5935,11 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
// disconnect our sync peer for stalling; we have bigger // disconnect our sync peer for stalling; we have bigger
// problems if we can't get any outbound peers. // problems if we can't get any outbound peers.
if (!pto->HasPermission(NetPermissionFlags::NoBan)) { if (!pto->HasPermission(NetPermissionFlags::NoBan)) {
LogPrintf("Timeout downloading headers from peer=%d, disconnecting\n", pto->GetId()); LogPrintf("Timeout downloading headers from peer=%d%s, disconnecting\n", pto->GetId(), fLogIPs ? strprintf(" peeraddr=%s", pto->addr.ToStringAddrPort()) : "");
pto->fDisconnect = true; pto->fDisconnect = true;
return true; return true;
} else { } else {
LogPrintf("Timeout downloading headers from noban peer=%d, not disconnecting\n", pto->GetId()); LogPrintf("Timeout downloading headers from noban peer=%d%s, not disconnecting\n", pto->GetId(), fLogIPs ? strprintf(" peeraddr=%s", pto->addr.ToStringAddrPort()) : "");
// Reset the headers sync state so that we have a // Reset the headers sync state so that we have a
// chance to try downloading from a different peer. // chance to try downloading from a different peer.
// Note: this will also result in at least one more // Note: this will also result in at least one more

View File

@ -230,6 +230,7 @@ static const CRPCConvertParam vRPCConvertParams[] =
{ "getnodeaddresses", 0, "count"}, { "getnodeaddresses", 0, "count"},
{ "addpeeraddress", 1, "port"}, { "addpeeraddress", 1, "port"},
{ "addpeeraddress", 2, "tried"}, { "addpeeraddress", 2, "tried"},
{ "sendmsgtopeer", 0, "peer_id" },
{ "stop", 0, "wait" }, { "stop", 0, "wait" },
{ "verifychainlock", 2, "blockHeight" }, { "verifychainlock", 2, "blockHeight" },
{ "verifyislock", 3, "maxHeight" }, { "verifyislock", 3, "maxHeight" },

View File

@ -352,7 +352,7 @@ static RPCHelpMan addconnection()
"\nOpen an outbound connection to a specified node. This RPC is for testing only.\n", "\nOpen an outbound connection to a specified node. This RPC is for testing only.\n",
{ {
{"address", RPCArg::Type::STR, RPCArg::Optional::NO, "The IP address and port to attempt connecting to."}, {"address", RPCArg::Type::STR, RPCArg::Optional::NO, "The IP address and port to attempt connecting to."},
{"connection_type", RPCArg::Type::STR, RPCArg::Optional::NO, "Type of connection to open (\"outbound-full-relay\", \"block-relay-only\" or \"addr-fetch\")."}, {"connection_type", RPCArg::Type::STR, RPCArg::Optional::NO, "Type of connection to open (\"outbound-full-relay\", \"block-relay-only\", \"addr-fetch\" or \"feeler\")."},
}, },
RPCResult{ RPCResult{
RPCResult::Type::OBJ, "", "", RPCResult::Type::OBJ, "", "",
@ -380,6 +380,8 @@ static RPCHelpMan addconnection()
conn_type = ConnectionType::BLOCK_RELAY; conn_type = ConnectionType::BLOCK_RELAY;
} else if (conn_type_in == "addr-fetch") { } else if (conn_type_in == "addr-fetch") {
conn_type = ConnectionType::ADDR_FETCH; conn_type = ConnectionType::ADDR_FETCH;
} else if (conn_type_in == "feeler") {
conn_type = ConnectionType::FEELER;
} else { } else {
throw JSONRPCError(RPC_INVALID_PARAMETER, self.ToString()); throw JSONRPCError(RPC_INVALID_PARAMETER, self.ToString());
} }
@ -1019,6 +1021,53 @@ static RPCHelpMan addpeeraddress()
}; };
} }
static RPCHelpMan sendmsgtopeer()
{
return RPCHelpMan{
"sendmsgtopeer",
"Send a p2p message to a peer specified by id.\n"
"The message type and body must be provided, the message header will be generated.\n"
"This RPC is for testing only.",
{
{"peer_id", RPCArg::Type::NUM, RPCArg::Optional::NO, "The peer to send the message to."},
{"msg_type", RPCArg::Type::STR, RPCArg::Optional::NO, strprintf("The message type (maximum length %i)", CMessageHeader::COMMAND_SIZE)},
{"msg", RPCArg::Type::STR_HEX, RPCArg::Optional::NO, "The serialized message body to send, in hex, without a message header"},
},
RPCResult{RPCResult::Type::NONE, "", ""},
RPCExamples{
HelpExampleCli("sendmsgtopeer", "0 \"addr\" \"ffffff\"") + HelpExampleRpc("sendmsgtopeer", "0 \"addr\" \"ffffff\"")},
[&](const RPCHelpMan& self, const JSONRPCRequest& request) -> UniValue {
const NodeId peer_id{request.params[0].get_int()};
const std::string& msg_type{request.params[1].get_str()};
if (msg_type.size() > CMessageHeader::COMMAND_SIZE) {
throw JSONRPCError(RPC_INVALID_PARAMETER, strprintf("Error: msg_type too long, max length is %i", CMessageHeader::COMMAND_SIZE));
}
const std::string& msg{request.params[2].get_str()};
if (!msg.empty() && !IsHex(msg)) {
throw JSONRPCError(RPC_INVALID_PARAMETER, "Error parsing input for msg");
}
NodeContext& node = EnsureAnyNodeContext(request.context);
CConnman& connman = EnsureConnman(node);
CSerializedNetMsg msg_ser;
msg_ser.data = ParseHex(msg);
msg_ser.m_type = msg_type;
bool success = connman.ForNode(peer_id, [&](CNode* node) {
connman.PushMessage(node, std::move(msg_ser));
return true;
});
if (!success) {
throw JSONRPCError(RPC_MISC_ERROR, "Error: Could not send message to peer");
}
return NullUniValue;
},
};
}
static RPCHelpMan setmnthreadactive() static RPCHelpMan setmnthreadactive()
{ {
return RPCHelpMan{"setmnthreadactive", return RPCHelpMan{"setmnthreadactive",
@ -1068,6 +1117,7 @@ static const CRPCCommand commands[] =
{ "hidden", &addconnection, }, { "hidden", &addconnection, },
{ "hidden", &addpeeraddress, }, { "hidden", &addpeeraddress, },
{ "hidden", &sendmsgtopeer },
{ "hidden", &setmnthreadactive }, { "hidden", &setmnthreadactive },
}; };
// clang-format on // clang-format on

View File

@ -149,6 +149,7 @@ const std::vector<std::string> RPC_COMMANDS_SAFE_FOR_FUZZING{
"pruneblockchain", "pruneblockchain",
"reconsiderblock", "reconsiderblock",
"scantxoutset", "scantxoutset",
"sendmsgtopeer", // when no peers are connected, no p2p message is sent
"sendrawtransaction", "sendrawtransaction",
"setmnthreadactive", "setmnthreadactive",
"setmocktime", "setmocktime",

View File

@ -6,8 +6,18 @@
from test_framework.p2p import P2PInterface from test_framework.p2p import P2PInterface
from test_framework.test_framework import BitcoinTestFramework from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import check_node_connections from test_framework.util import (
assert_equal,
check_node_connections,
)
class P2PFeelerReceiver(P2PInterface):
def on_version(self, message):
# The bitcoind node closes feeler connections as soon as a version
# message is received from the test framework. Don't send any responses
# to the node's version message since the connection will already be
# closed.
pass
class P2PAddConnections(BitcoinTestFramework): class P2PAddConnections(BitcoinTestFramework):
def set_test_params(self): def set_test_params(self):
@ -86,6 +96,16 @@ class P2PAddConnections(BitcoinTestFramework):
check_node_connections(node=self.nodes[1], num_in=5, num_out=10) check_node_connections(node=self.nodes[1], num_in=5, num_out=10)
self.log.info("Add 1 feeler connection to node 0")
feeler_conn = self.nodes[0].add_outbound_p2p_connection(P2PFeelerReceiver(), p2p_idx=6, connection_type="feeler")
# Feeler connection is closed
assert not feeler_conn.is_connected
# Verify version message received
assert_equal(feeler_conn.message_count["version"], 1)
# Feeler connections do not request tx relay
assert_equal(feeler_conn.last_message["version"].relay, 0)
if __name__ == '__main__': if __name__ == '__main__':
P2PAddConnections().main() P2PAddConnections().main()

View File

@ -91,7 +91,7 @@ class DisconnectBanTest(BitcoinTestFramework):
self.log.info("disconnectnode: successfully disconnect node by address") self.log.info("disconnectnode: successfully disconnect node by address")
address1 = self.nodes[0].getpeerinfo()[0]['addr'] address1 = self.nodes[0].getpeerinfo()[0]['addr']
self.nodes[0].disconnectnode(address=address1) self.nodes[0].disconnectnode(address=address1)
self.wait_until(lambda: len(self.nodes[0].getpeerinfo()) == 1, timeout=10) self.wait_until(lambda: len(self.nodes[1].getpeerinfo()) == 1, timeout=10)
assert not [node for node in self.nodes[0].getpeerinfo() if node['addr'] == address1] assert not [node for node in self.nodes[0].getpeerinfo() if node['addr'] == address1]
self.log.info("disconnectnode: successfully reconnect node") self.log.info("disconnectnode: successfully reconnect node")
@ -102,7 +102,7 @@ class DisconnectBanTest(BitcoinTestFramework):
self.log.info("disconnectnode: successfully disconnect node by node id") self.log.info("disconnectnode: successfully disconnect node by node id")
id1 = self.nodes[0].getpeerinfo()[0]['id'] id1 = self.nodes[0].getpeerinfo()[0]['id']
self.nodes[0].disconnectnode(nodeid=id1) self.nodes[0].disconnectnode(nodeid=id1)
self.wait_until(lambda: len(self.nodes[0].getpeerinfo()) == 1, timeout=10) self.wait_until(lambda: len(self.nodes[1].getpeerinfo()) == 1, timeout=10)
assert not [node for node in self.nodes[0].getpeerinfo() if node['id'] == id1] assert not [node for node in self.nodes[0].getpeerinfo() if node['id'] == id1]
if __name__ == '__main__': if __name__ == '__main__':

View File

@ -0,0 +1,37 @@
#!/usr/bin/env python3
# Copyright (c) 2023-present The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
import threading
from test_framework.messages import MAX_PROTOCOL_MESSAGE_LENGTH
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import random_bytes
class NetDeadlockTest(BitcoinTestFramework):
def set_test_params(self):
self.setup_clean_chain = True
self.num_nodes = 2
def run_test(self):
node0 = self.nodes[0]
node1 = self.nodes[1]
self.log.info("Simultaneously send a large message on both sides")
rand_msg = random_bytes(MAX_PROTOCOL_MESSAGE_LENGTH).hex()
thread0 = threading.Thread(target=node0.sendmsgtopeer, args=(0, "unknown", rand_msg))
thread1 = threading.Thread(target=node1.sendmsgtopeer, args=(0, "unknown", rand_msg))
thread0.start()
thread1.start()
thread0.join()
thread1.join()
self.log.info("Check whether a deadlock happened")
self.nodes[0].generate(1)
self.sync_blocks()
if __name__ == '__main__':
NetDeadlockTest().main()

View File

@ -10,6 +10,7 @@ Tests correspond to code in rpc/net.cpp.
from test_framework.p2p import P2PInterface from test_framework.p2p import P2PInterface
import test_framework.messages import test_framework.messages
from test_framework.messages import ( from test_framework.messages import (
MAX_PROTOCOL_MESSAGE_LENGTH,
NODE_NETWORK, NODE_NETWORK,
) )
@ -66,6 +67,7 @@ class NetTest(DashTestFramework):
self.test_service_flags() self.test_service_flags()
self.test_getnodeaddresses() self.test_getnodeaddresses()
self.test_addpeeraddress() self.test_addpeeraddress()
self.test_sendmsgtopeer()
def test_connection_count(self): def test_connection_count(self):
self.log.info("Test getconnectioncount") self.log.info("Test getconnectioncount")
@ -341,6 +343,37 @@ class NetTest(DashTestFramework):
addrs = node.getnodeaddresses(count=0) # getnodeaddresses re-runs the addrman checks addrs = node.getnodeaddresses(count=0) # getnodeaddresses re-runs the addrman checks
assert_equal(len(addrs), 2) assert_equal(len(addrs), 2)
def test_sendmsgtopeer(self):
node = self.nodes[0]
self.restart_node(0)
self.connect_nodes(0, 1)
self.log.info("Test sendmsgtopeer")
self.log.debug("Send a valid message")
with self.nodes[1].assert_debug_log(expected_msgs=["received: addr"]):
node.sendmsgtopeer(peer_id=0, msg_type="addr", msg="FFFFFF")
self.log.debug("Test error for sending to non-existing peer")
assert_raises_rpc_error(-1, "Error: Could not send message to peer", node.sendmsgtopeer, peer_id=100, msg_type="addr", msg="FF")
self.log.debug("Test that zero-length msg_type is allowed")
node.sendmsgtopeer(peer_id=0, msg_type="addr", msg="")
self.log.debug("Test error for msg_type that is too long")
assert_raises_rpc_error(-8, "Error: msg_type too long, max length is 12", node.sendmsgtopeer, peer_id=0, msg_type="long_msg_type", msg="FF")
self.log.debug("Test that unknown msg_type is allowed")
node.sendmsgtopeer(peer_id=0, msg_type="unknown", msg="FF")
self.log.debug("Test that empty msg is allowed")
node.sendmsgtopeer(peer_id=0, msg_type="addr", msg="FF")
self.log.debug("Test that oversized messages are allowed, but get us disconnected")
zero_byte_string = b'\x00' * int(MAX_PROTOCOL_MESSAGE_LENGTH + 1)
node.sendmsgtopeer(peer_id=0, msg_type="addr", msg=zero_byte_string.hex())
self.wait_until(lambda: len(self.nodes[0].getpeerinfo()) == 0, timeout=10)
if __name__ == '__main__': if __name__ == '__main__':
NetTest().main() NetTest().main()

View File

@ -695,42 +695,61 @@ class BitcoinTestFramework(metaclass=BitcoinTestMetaClass):
if (a == b): if (a == b):
return return
def connect_nodes_helper(from_connection, node_num): from_connection = self.nodes[a]
ip_port = "127.0.0.1:" + str(p2p_port(node_num)) to_connection = self.nodes[b]
from_connection.addnode(ip_port, "onetry") ip_port = "127.0.0.1:" + str(p2p_port(b))
# poll until version handshake complete to avoid race conditions from_connection.addnode(ip_port, "onetry")
# with transaction relaying
# See comments in net_processing:
# * Must have a version message before anything else
# * Must have a verack message before anything else
wait_until_helper(lambda: all(peer['version'] != 0 for peer in from_connection.getpeerinfo()))
wait_until_helper(lambda: all(peer['bytesrecv_per_msg'].pop('verack', 0) == 24 for peer in from_connection.getpeerinfo()))
connect_nodes_helper(self.nodes[a], b) # Use subversion as peer id. Test nodes have their node number appended to the user agent string
from_connection_subver = from_connection.getnetworkinfo()['subversion']
to_connection_subver = to_connection.getnetworkinfo()['subversion']
def find_conn(node, peer_subversion, inbound):
return next(filter(lambda peer: peer['subver'] == peer_subversion and peer['inbound'] == inbound, node.getpeerinfo()), None)
# poll until version handshake complete to avoid race conditions
# with transaction relaying
# See comments in net_processing:
# * Must have a version message before anything else
# * Must have a verack message before anything else
self.wait_until(lambda: find_conn(from_connection, to_connection_subver, inbound=False) is not None)
self.wait_until(lambda: find_conn(to_connection, from_connection_subver, inbound=True) is not None)
def check_bytesrecv(peer, msg_type, min_bytes_recv):
assert peer is not None, "Error: peer disconnected"
return peer['bytesrecv_per_msg'].pop(msg_type, 0) >= min_bytes_recv
self.wait_until(lambda: check_bytesrecv(find_conn(from_connection, to_connection_subver, inbound=False), 'verack', 24))
self.wait_until(lambda: check_bytesrecv(find_conn(to_connection, from_connection_subver, inbound=True), 'verack', 24))
# The message bytes are counted before processing the message, so make
# sure it was fully processed by waiting for a ping.
self.wait_until(lambda: check_bytesrecv(find_conn(from_connection, to_connection_subver, inbound=False), 'pong', 32))
self.wait_until(lambda: check_bytesrecv(find_conn(to_connection, from_connection_subver, inbound=True), 'pong', 32))
def disconnect_nodes(self, a, b): def disconnect_nodes(self, a, b):
# A node cannot disconnect from itself, bail out early # A node cannot disconnect from itself, bail out early
if (a == b): if (a == b):
return return
def disconnect_nodes_helper(from_connection, node_num): def disconnect_nodes_helper(node_a, node_b):
def get_peer_ids(): def get_peer_ids(from_connection, node_num):
result = [] result = []
for peer in from_connection.getpeerinfo(): for peer in from_connection.getpeerinfo():
if "testnode{}".format(node_num) in peer['subver']: if "testnode{}".format(node_num) in peer['subver']:
result.append(peer['id']) result.append(peer['id'])
return result return result
peer_ids = get_peer_ids() peer_ids = get_peer_ids(node_a, node_b.index)
if not peer_ids: if not peer_ids:
self.log.warning("disconnect_nodes: {} and {} were not connected".format( self.log.warning("disconnect_nodes: {} and {} were not connected".format(
from_connection.index, node_a.index,
node_num, node_b.index,
)) ))
return return
for peer_id in peer_ids: for peer_id in peer_ids:
try: try:
from_connection.disconnectnode(nodeid=peer_id) node_a.disconnectnode(nodeid=peer_id)
except JSONRPCException as e: except JSONRPCException as e:
# If this node is disconnected between calculating the peer id # If this node is disconnected between calculating the peer id
# and issuing the disconnect, don't worry about it. # and issuing the disconnect, don't worry about it.
@ -739,9 +758,10 @@ class BitcoinTestFramework(metaclass=BitcoinTestMetaClass):
raise raise
# wait to disconnect # wait to disconnect
wait_until_helper(lambda: not get_peer_ids(), timeout=5) self.wait_until(lambda: not get_peer_ids(node_a, node_b.index), timeout=5)
self.wait_until(lambda: not get_peer_ids(node_b, node_a.index), timeout=5)
disconnect_nodes_helper(self.nodes[a], b) disconnect_nodes_helper(self.nodes[a], self.nodes[b])
def isolate_node(self, node_num, timeout=5): def isolate_node(self, node_num, timeout=5):
self.nodes[node_num].setnetworkactive(False) self.nodes[node_num].setnetworkactive(False)

View File

@ -103,7 +103,7 @@ class TestNode():
"-debug", "-debug",
"-debugexclude=libevent", "-debugexclude=libevent",
"-debugexclude=leveldb", "-debugexclude=leveldb",
"-uacomment=testnode%d" % i, "-uacomment=testnode%d" % i, # required for subversion uniqueness across peers
] ]
if self.mocktime != 0: if self.mocktime != 0:
self.args.append(f"-mocktime={mocktime}") self.args.append(f"-mocktime={mocktime}")
@ -574,7 +574,7 @@ class TestNode():
def add_outbound_p2p_connection(self, p2p_conn, *, p2p_idx, connection_type="outbound-full-relay", **kwargs): def add_outbound_p2p_connection(self, p2p_conn, *, p2p_idx, connection_type="outbound-full-relay", **kwargs):
"""Add an outbound p2p connection from node. Must be an """Add an outbound p2p connection from node. Must be an
"outbound-full-relay", "block-relay-only" or "addr-fetch" connection. "outbound-full-relay", "block-relay-only", "addr-fetch" or "feeler" connection.
This method adds the p2p connection to the self.p2ps list and returns This method adds the p2p connection to the self.p2ps list and returns
the connection to the caller. the connection to the caller.
@ -586,11 +586,16 @@ class TestNode():
p2p_conn.peer_accept_connection(connect_cb=addconnection_callback, connect_id=p2p_idx + 1, net=self.chain, timeout_factor=self.timeout_factor, **kwargs)() p2p_conn.peer_accept_connection(connect_cb=addconnection_callback, connect_id=p2p_idx + 1, net=self.chain, timeout_factor=self.timeout_factor, **kwargs)()
p2p_conn.wait_for_connect() if connection_type == "feeler":
self.p2ps.append(p2p_conn) # feeler connections are closed as soon as the node receives a `version` message
p2p_conn.wait_until(lambda: p2p_conn.message_count["version"] == 1, check_connected=False)
p2p_conn.wait_until(lambda: not p2p_conn.is_connected, check_connected=False)
else:
p2p_conn.wait_for_connect()
self.p2ps.append(p2p_conn)
p2p_conn.wait_for_verack() p2p_conn.wait_for_verack()
p2p_conn.sync_with_ping() p2p_conn.sync_with_ping()
return p2p_conn return p2p_conn

View File

@ -13,6 +13,7 @@ import inspect
import json import json
import logging import logging
import os import os
import random
import shutil import shutil
import re import re
import time import time
@ -274,6 +275,11 @@ def sha256sum_file(filename):
d = f.read(4096) d = f.read(4096)
return h.digest() return h.digest()
# TODO: Remove and use random.randbytes(n) instead, available in Python 3.9
def random_bytes(n):
"""Return a random bytes object of length n."""
return bytes(random.getrandbits(8) for i in range(n))
# RPC/P2P connection constants and functions # RPC/P2P connection constants and functions
############################################ ############################################

View File

@ -259,6 +259,7 @@ BASE_SCRIPTS = [
'p2p_leak_tx.py', 'p2p_leak_tx.py',
'p2p_eviction.py', 'p2p_eviction.py',
'p2p_ibd_stalling.py', 'p2p_ibd_stalling.py',
'p2p_net_deadlock.py',
'rpc_signmessage.py', 'rpc_signmessage.py',
'rpc_generateblock.py', 'rpc_generateblock.py',
'rpc_generate.py', 'rpc_generate.py',