foster/bitcoin-bitcoin-core
0
0
Fork 0
mirror of https://github.com/bitcoin/bitcoin.git synced 2025-02-02 09:46:52 -05:00
Code Issues Activity

[net_processing] Move ProcessMessage to PeerLogicValidation

Browse source
This commit is contained in:
John Newbery 2020-08-12 12:13:53 +01:00
parent c556770b5e
commit daed542a12
3 changed files with 67 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

120
src/net_processing.cpp
View file

@ -2303,17 +2303,9 @@ static void ProcessGetCFCheckPt(CNode& pfrom, CDataStream& vRecv, const CChainPa
connman.PushMessage(&pfrom, std::move(msg)); connman.PushMessage(&pfrom, std::move(msg));
} }
void ProcessMessage( void PeerLogicValidation::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv,
CNode& pfrom,
const std::string& msg_type,
CDataStream& vRecv,
const std::chrono::microseconds time_received, const std::chrono::microseconds time_received,
const CChainParams& chainparams, const CChainParams& chainparams, const std::atomic<bool>& interruptMsgProc)
ChainstateManager& chainman,
CTxMemPool& mempool,
CConnman& connman,
BanMan* banman,
const std::atomic<bool>& interruptMsgProc)
{ {
LogPrint(BCLog::NET, "received: %s (%u bytes) peer=%d\n", SanitizeString(msg_type), vRecv.size(), pfrom.GetId()); LogPrint(BCLog::NET, "received: %s (%u bytes) peer=%d\n", SanitizeString(msg_type), vRecv.size(), pfrom.GetId());
if (gArgs.IsArgSet("-dropmessagestest") && GetRand(gArgs.GetArg("-dropmessagestest", 0)) == 0) if (gArgs.IsArgSet("-dropmessagestest") && GetRand(gArgs.GetArg("-dropmessagestest", 0)) == 0)
@ -2349,7 +2341,7 @@ void ProcessMessage(
nServices = ServiceFlags(nServiceInt); nServices = ServiceFlags(nServiceInt);
if (!pfrom.IsInboundConn()) if (!pfrom.IsInboundConn())
{ {
connman.SetServices(pfrom.addr, nServices); m_connman.SetServices(pfrom.addr, nServices);
} }
if (pfrom.ExpectServicesFromConn() && !HasAllDesirableServiceFlags(nServices)) if (pfrom.ExpectServicesFromConn() && !HasAllDesirableServiceFlags(nServices))
{ {
@ -2378,7 +2370,7 @@ void ProcessMessage(
if (!vRecv.empty()) if (!vRecv.empty())
vRecv >> fRelay; vRecv >> fRelay;
// Disconnect if we connected to ourself // Disconnect if we connected to ourself
if (pfrom.IsInboundConn() && !connman.CheckIncomingNonce(nNonce)) if (pfrom.IsInboundConn() && !m_connman.CheckIncomingNonce(nNonce))
{ {
LogPrintf("connected to self at %s, disconnecting\n", pfrom.addr.ToString()); LogPrintf("connected to self at %s, disconnecting\n", pfrom.addr.ToString());
pfrom.fDisconnect = true; pfrom.fDisconnect = true;
@ -2392,13 +2384,13 @@ void ProcessMessage(
// Be shy and don't send version until we hear // Be shy and don't send version until we hear
if (pfrom.IsInboundConn()) if (pfrom.IsInboundConn())
PushNodeVersion(pfrom, connman, GetAdjustedTime()); PushNodeVersion(pfrom, m_connman, GetAdjustedTime());
if (nVersion >= WTXID_RELAY_VERSION) { if (nVersion >= WTXID_RELAY_VERSION) {
connman.PushMessage(&pfrom, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::WTXIDRELAY)); m_connman.PushMessage(&pfrom, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::WTXIDRELAY));
} }
connman.PushMessage(&pfrom, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::VERACK)); m_connman.PushMessage(&pfrom, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::VERACK));
pfrom.nServices = nServices; pfrom.nServices = nServices;
pfrom.SetAddrLocal(addrMe); pfrom.SetAddrLocal(addrMe);
@ -2454,9 +2446,9 @@ void ProcessMessage(
} }
// Get recent addresses // Get recent addresses
connman.PushMessage(&pfrom, CNetMsgMaker(nSendVersion).Make(NetMsgType::GETADDR)); m_connman.PushMessage(&pfrom, CNetMsgMaker(nSendVersion).Make(NetMsgType::GETADDR));
pfrom.fGetAddr = true; pfrom.fGetAddr = true;
connman.MarkAddressGood(pfrom.addr); m_connman.MarkAddressGood(pfrom.addr);
} }
std::string remoteAddr; std::string remoteAddr;
@ -2475,7 +2467,7 @@ void ProcessMessage(
// If the peer is old enough to have the old alert system, send it the final alert. // If the peer is old enough to have the old alert system, send it the final alert.
if (pfrom.nVersion <= 70012) { if (pfrom.nVersion <= 70012) {
CDataStream finalAlert(ParseHex("60010000000000000000000000ffffff7f00000000ffffff7ffeffff7f01ffffff7f00000000ffffff7f00ffffff7f002f555247454e543a20416c657274206b657920636f6d70726f6d697365642c2075706772616465207265717569726564004630440220653febd6410f470f6bae11cad19c48413becb1ac2c17f908fd0fd53bdc3abd5202206d0e9c96fe88d4a0f01ed9dedae2b6f9e00da94cad0fecaae66ecf689bf71b50"), SER_NETWORK, PROTOCOL_VERSION); CDataStream finalAlert(ParseHex("60010000000000000000000000ffffff7f00000000ffffff7ffeffff7f01ffffff7f00000000ffffff7f00ffffff7f002f555247454e543a20416c657274206b657920636f6d70726f6d697365642c2075706772616465207265717569726564004630440220653febd6410f470f6bae11cad19c48413becb1ac2c17f908fd0fd53bdc3abd5202206d0e9c96fe88d4a0f01ed9dedae2b6f9e00da94cad0fecaae66ecf689bf71b50"), SER_NETWORK, PROTOCOL_VERSION);
connman.PushMessage(&pfrom, CNetMsgMaker(nSendVersion).Make("alert", finalAlert)); m_connman.PushMessage(&pfrom, CNetMsgMaker(nSendVersion).Make("alert", finalAlert));
} }
// Feeler connections exist only to verify if address is online. // Feeler connections exist only to verify if address is online.
@ -2514,7 +2506,7 @@ void ProcessMessage(
// We send this to non-NODE NETWORK peers as well, because even // We send this to non-NODE NETWORK peers as well, because even
// non-NODE NETWORK peers can announce blocks (such as pruning // non-NODE NETWORK peers can announce blocks (such as pruning
// nodes) // nodes)
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::SENDHEADERS)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::SENDHEADERS));
} }
if (pfrom.nVersion >= SHORT_IDS_BLOCKS_VERSION) { if (pfrom.nVersion >= SHORT_IDS_BLOCKS_VERSION) {
// Tell our peer we are willing to provide version 1 or 2 cmpctblocks // Tell our peer we are willing to provide version 1 or 2 cmpctblocks
@ -2525,9 +2517,9 @@ void ProcessMessage(
bool fAnnounceUsingCMPCTBLOCK = false; bool fAnnounceUsingCMPCTBLOCK = false;
uint64_t nCMPCTBLOCKVersion = 2; uint64_t nCMPCTBLOCKVersion = 2;
if (pfrom.GetLocalServices() & NODE_WITNESS) if (pfrom.GetLocalServices() & NODE_WITNESS)
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::SENDCMPCT, fAnnounceUsingCMPCTBLOCK, nCMPCTBLOCKVersion)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::SENDCMPCT, fAnnounceUsingCMPCTBLOCK, nCMPCTBLOCKVersion));
nCMPCTBLOCKVersion = 1; nCMPCTBLOCKVersion = 1;
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::SENDCMPCT, fAnnounceUsingCMPCTBLOCK, nCMPCTBLOCKVersion)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::SENDCMPCT, fAnnounceUsingCMPCTBLOCK, nCMPCTBLOCKVersion));
} }
pfrom.fSuccessfullyConnected = true; pfrom.fSuccessfullyConnected = true;
return; return;
@ -2591,7 +2583,7 @@ void ProcessMessage(
if (addr.nTime <= 100000000 || addr.nTime > nNow + 10 * 60) if (addr.nTime <= 100000000 || addr.nTime > nNow + 10 * 60)
addr.nTime = nNow - 5 * 24 * 60 * 60; addr.nTime = nNow - 5 * 24 * 60 * 60;
pfrom.AddAddressKnown(addr); pfrom.AddAddressKnown(addr);
if (banman && (banman->IsDiscouraged(addr) || banman->IsBanned(addr))) { if (m_banman && (m_banman->IsDiscouraged(addr) || m_banman->IsBanned(addr))) {
// Do not process banned/discouraged addresses beyond remembering we received them // Do not process banned/discouraged addresses beyond remembering we received them
continue; continue;
} }
@ -2599,13 +2591,13 @@ void ProcessMessage(
if (addr.nTime > nSince && !pfrom.fGetAddr && vAddr.size() <= 10 && addr.IsRoutable()) if (addr.nTime > nSince && !pfrom.fGetAddr && vAddr.size() <= 10 && addr.IsRoutable())
{ {
// Relay to a limited number of other nodes // Relay to a limited number of other nodes
RelayAddress(addr, fReachable, connman); RelayAddress(addr, fReachable, m_connman);
} }
// Do not store addresses outside our network // Do not store addresses outside our network
if (fReachable) if (fReachable)
vAddrOk.push_back(addr); vAddrOk.push_back(addr);
} }
connman.AddNewAddresses(vAddrOk, pfrom.addr, 2 * 60 * 60); m_connman.AddNewAddresses(vAddrOk, pfrom.addr, 2 * 60 * 60);
if (vAddr.size() < 1000) if (vAddr.size() < 1000)
pfrom.fGetAddr = false; pfrom.fGetAddr = false;
if (pfrom.IsAddrFetchConn()) if (pfrom.IsAddrFetchConn())
@ -2681,7 +2673,7 @@ void ProcessMessage(
if (inv.IsMsgWtx()) continue; if (inv.IsMsgWtx()) continue;
} }
bool fAlreadyHave = AlreadyHave(inv, mempool); bool fAlreadyHave = AlreadyHave(inv, m_mempool);
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());
if (inv.IsMsgTx()) { if (inv.IsMsgTx()) {
@ -2704,14 +2696,14 @@ void ProcessMessage(
LogPrint(BCLog::NET, "transaction (%s) inv sent in violation of protocol, disconnecting peer=%d\n", inv.hash.ToString(), pfrom.GetId()); LogPrint(BCLog::NET, "transaction (%s) inv sent in violation of protocol, disconnecting peer=%d\n", inv.hash.ToString(), pfrom.GetId());
pfrom.fDisconnect = true; pfrom.fDisconnect = true;
return; return;
} else if (!fAlreadyHave && !chainman.ActiveChainstate().IsInitialBlockDownload()) { } else if (!fAlreadyHave && !m_chainman.ActiveChainstate().IsInitialBlockDownload()) {
RequestTx(State(pfrom.GetId()), ToGenTxid(inv), current_time); RequestTx(State(pfrom.GetId()), ToGenTxid(inv), current_time);
} }
} }
} }
if (best_block != nullptr) { if (best_block != nullptr) {
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(pindexBestHeader), *best_block)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(pindexBestHeader), *best_block));
LogPrint(BCLog::NET, "getheaders (%d) %s to peer=%d\n", pindexBestHeader->nHeight, best_block->ToString(), pfrom.GetId()); LogPrint(BCLog::NET, "getheaders (%d) %s to peer=%d\n", pindexBestHeader->nHeight, best_block->ToString(), pfrom.GetId());
} }
@ -2735,7 +2727,7 @@ void ProcessMessage(
} }
pfrom.vRecvGetData.insert(pfrom.vRecvGetData.end(), vInv.begin(), vInv.end()); pfrom.vRecvGetData.insert(pfrom.vRecvGetData.end(), vInv.begin(), vInv.end());
ProcessGetData(pfrom, chainparams, connman, mempool, interruptMsgProc); ProcessGetData(pfrom, chainparams, m_connman, m_mempool, interruptMsgProc);
return; return;
} }
@ -2819,7 +2811,7 @@ void ProcessMessage(
// Unlock cs_most_recent_block to avoid cs_main lock inversion // Unlock cs_most_recent_block to avoid cs_main lock inversion
} }
if (recent_block) { if (recent_block) {
SendBlockTransactions(*recent_block, req, pfrom, connman); SendBlockTransactions(*recent_block, req, pfrom, m_connman);
return; return;
} }
@ -2852,7 +2844,7 @@ void ProcessMessage(
bool ret = ReadBlockFromDisk(block, pindex, chainparams.GetConsensus()); bool ret = ReadBlockFromDisk(block, pindex, chainparams.GetConsensus());
assert(ret); assert(ret);
SendBlockTransactions(block, req, pfrom, connman); SendBlockTransactions(block, req, pfrom, m_connman);
return; return;
} }
@ -2919,7 +2911,7 @@ void ProcessMessage(
// will re-announce the new block via headers (or compact blocks again) // will re-announce the new block via headers (or compact blocks again)
// in the SendMessages logic. // in the SendMessages logic.
nodestate->pindexBestHeaderSent = pindex ? pindex : ::ChainActive().Tip(); nodestate->pindexBestHeaderSent = pindex ? pindex : ::ChainActive().Tip();
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::HEADERS, vHeaders)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::HEADERS, vHeaders));
return; return;
} }
@ -2978,10 +2970,10 @@ void ProcessMessage(
// already; and an adversary can already relay us old transactions // already; and an adversary can already relay us old transactions
// (older than our recency filter) if trying to DoS us, without any need // (older than our recency filter) if trying to DoS us, without any need
// for witness malleation. // for witness malleation.
if (!AlreadyHave(CInv(MSG_WTX, wtxid), mempool) && if (!AlreadyHave(CInv(MSG_WTX, wtxid), m_mempool) &&
AcceptToMemoryPool(mempool, state, ptx, &lRemovedTxn, false /* bypass_limits */, 0 /* nAbsurdFee */)) { AcceptToMemoryPool(m_mempool, state, ptx, &lRemovedTxn, false /* bypass_limits */, 0 /* nAbsurdFee */)) {
mempool.check(&::ChainstateActive().CoinsTip()); m_mempool.check(&::ChainstateActive().CoinsTip());
RelayTransaction(tx.GetHash(), tx.GetWitnessHash(), connman); RelayTransaction(tx.GetHash(), tx.GetWitnessHash(), m_connman);
for (unsigned int i = 0; i < tx.vout.size(); i++) { for (unsigned int i = 0; i < tx.vout.size(); i++) {
auto it_by_prev = mapOrphanTransactionsByPrev.find(COutPoint(txid, i)); auto it_by_prev = mapOrphanTransactionsByPrev.find(COutPoint(txid, i));
if (it_by_prev != mapOrphanTransactionsByPrev.end()) { if (it_by_prev != mapOrphanTransactionsByPrev.end()) {
@ -2996,10 +2988,10 @@ void ProcessMessage(
LogPrint(BCLog::MEMPOOL, "AcceptToMemoryPool: peer=%d: accepted %s (poolsz %u txn, %u kB)\n", LogPrint(BCLog::MEMPOOL, "AcceptToMemoryPool: peer=%d: accepted %s (poolsz %u txn, %u kB)\n",
pfrom.GetId(), pfrom.GetId(),
tx.GetHash().ToString(), tx.GetHash().ToString(),
mempool.size(), mempool.DynamicMemoryUsage() / 1000); m_mempool.size(), m_mempool.DynamicMemoryUsage() / 1000);
// Recursively process any orphan transactions that depended on this one // Recursively process any orphan transactions that depended on this one
ProcessOrphanTx(connman, mempool, pfrom.orphan_work_set, lRemovedTxn); ProcessOrphanTx(m_connman, m_mempool, pfrom.orphan_work_set, lRemovedTxn);
} }
else if (state.GetResult() == TxValidationResult::TX_MISSING_INPUTS) else if (state.GetResult() == TxValidationResult::TX_MISSING_INPUTS)
{ {
@ -3033,7 +3025,7 @@ void ProcessMessage(
// protocol for getting all unconfirmed parents. // protocol for getting all unconfirmed parents.
CInv _inv(MSG_TX | nFetchFlags, parent_txid); CInv _inv(MSG_TX | nFetchFlags, parent_txid);
pfrom.AddKnownTx(parent_txid); pfrom.AddKnownTx(parent_txid);
if (!AlreadyHave(_inv, mempool)) RequestTx(State(pfrom.GetId()), ToGenTxid(_inv), current_time); if (!AlreadyHave(_inv, m_mempool)) RequestTx(State(pfrom.GetId()), ToGenTxid(_inv), current_time);
} }
AddOrphanTx(ptx, pfrom.GetId()); AddOrphanTx(ptx, pfrom.GetId());
@ -3094,11 +3086,11 @@ void ProcessMessage(
// if they were already in the mempool, // if they were already in the mempool,
// allowing the node to function as a gateway for // allowing the node to function as a gateway for
// nodes hidden behind it. // nodes hidden behind it.
if (!mempool.exists(tx.GetHash())) { if (!m_mempool.exists(tx.GetHash())) {
LogPrintf("Not relaying non-mempool transaction %s from forcerelay peer=%d\n", tx.GetHash().ToString(), pfrom.GetId()); LogPrintf("Not relaying non-mempool transaction %s from forcerelay peer=%d\n", tx.GetHash().ToString(), pfrom.GetId());
} else { } else {
LogPrintf("Force relaying tx %s from peer=%d\n", tx.GetHash().ToString(), pfrom.GetId()); LogPrintf("Force relaying tx %s from peer=%d\n", tx.GetHash().ToString(), pfrom.GetId());
RelayTransaction(tx.GetHash(), tx.GetWitnessHash(), connman); RelayTransaction(tx.GetHash(), tx.GetWitnessHash(), m_connman);
} }
} }
} }
@ -3151,7 +3143,7 @@ void ProcessMessage(
if (!LookupBlockIndex(cmpctblock.header.hashPrevBlock)) { if (!LookupBlockIndex(cmpctblock.header.hashPrevBlock)) {
// Doesn't connect (or is genesis), instead of DoSing in AcceptBlockHeader, request deeper headers // Doesn't connect (or is genesis), instead of DoSing in AcceptBlockHeader, request deeper headers
if (!::ChainstateActive().IsInitialBlockDownload()) if (!::ChainstateActive().IsInitialBlockDownload())
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(pindexBestHeader), uint256())); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(pindexBestHeader), uint256()));
return; return;
} }
@ -3162,7 +3154,7 @@ void ProcessMessage(
const CBlockIndex *pindex = nullptr; const CBlockIndex *pindex = nullptr;
BlockValidationState state; BlockValidationState state;
if (!chainman.ProcessNewBlockHeaders({cmpctblock.header}, state, chainparams, &pindex)) { if (!m_chainman.ProcessNewBlockHeaders({cmpctblock.header}, state, chainparams, &pindex)) {
if (state.IsInvalid()) { if (state.IsInvalid()) {
MaybePunishNodeForBlock(pfrom.GetId(), state, /*via_compact_block*/ true, "invalid header via cmpctblock"); MaybePunishNodeForBlock(pfrom.GetId(), state, /*via_compact_block*/ true, "invalid header via cmpctblock");
return; return;
@ -3212,7 +3204,7 @@ void ProcessMessage(
// so we just grab the block via normal getdata // so we just grab the block via normal getdata
std::vector<CInv> vInv(1); std::vector<CInv> vInv(1);
vInv[0] = CInv(MSG_BLOCK | GetFetchFlags(pfrom), cmpctblock.header.GetHash()); vInv[0] = CInv(MSG_BLOCK | GetFetchFlags(pfrom), cmpctblock.header.GetHash());
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETDATA, vInv)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETDATA, vInv));
} }
return; return;
} }
@ -3233,9 +3225,9 @@ void ProcessMessage(
if ((!fAlreadyInFlight && nodestate->nBlocksInFlight < MAX_BLOCKS_IN_TRANSIT_PER_PEER) || if ((!fAlreadyInFlight && nodestate->nBlocksInFlight < MAX_BLOCKS_IN_TRANSIT_PER_PEER) ||
(fAlreadyInFlight && blockInFlightIt->second.first == pfrom.GetId())) { (fAlreadyInFlight && blockInFlightIt->second.first == pfrom.GetId())) {
std::list<QueuedBlock>::iterator* queuedBlockIt = nullptr; std::list<QueuedBlock>::iterator* queuedBlockIt = nullptr;
if (!MarkBlockAsInFlight(mempool, pfrom.GetId(), pindex->GetBlockHash(), pindex, &queuedBlockIt)) { if (!MarkBlockAsInFlight(m_mempool, pfrom.GetId(), pindex->GetBlockHash(), pindex, &queuedBlockIt)) {
if (!(*queuedBlockIt)->partialBlock) if (!(*queuedBlockIt)->partialBlock)
(*queuedBlockIt)->partialBlock.reset(new PartiallyDownloadedBlock(&mempool)); (*queuedBlockIt)->partialBlock.reset(new PartiallyDownloadedBlock(&m_mempool));
else { else {
// The block was already in flight using compact blocks from the same peer // The block was already in flight using compact blocks from the same peer
LogPrint(BCLog::NET, "Peer sent us compact block we were already syncing!\n"); LogPrint(BCLog::NET, "Peer sent us compact block we were already syncing!\n");
@ -3253,7 +3245,7 @@ void ProcessMessage(
// Duplicate txindexes, the block is now in-flight, so just request it // Duplicate txindexes, the block is now in-flight, so just request it
std::vector<CInv> vInv(1); std::vector<CInv> vInv(1);
vInv[0] = CInv(MSG_BLOCK | GetFetchFlags(pfrom), cmpctblock.header.GetHash()); vInv[0] = CInv(MSG_BLOCK | GetFetchFlags(pfrom), cmpctblock.header.GetHash());
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETDATA, vInv)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETDATA, vInv));
return; return;
} }
@ -3270,7 +3262,7 @@ void ProcessMessage(
fProcessBLOCKTXN = true; fProcessBLOCKTXN = true;
} else { } else {
req.blockhash = pindex->GetBlockHash(); req.blockhash = pindex->GetBlockHash();
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETBLOCKTXN, req)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETBLOCKTXN, req));
} }
} else { } else {
// This block is either already in flight from a different // This block is either already in flight from a different
@ -3278,7 +3270,7 @@ void ProcessMessage(
// download from. // download from.
// Optimistically try to reconstruct anyway since we might be // Optimistically try to reconstruct anyway since we might be
// able to without any round trips. // able to without any round trips.
PartiallyDownloadedBlock tempBlock(&mempool); PartiallyDownloadedBlock tempBlock(&m_mempool);
ReadStatus status = tempBlock.InitData(cmpctblock, vExtraTxnForCompact); ReadStatus status = tempBlock.InitData(cmpctblock, vExtraTxnForCompact);
if (status != READ_STATUS_OK) { if (status != READ_STATUS_OK) {
// TODO: don't ignore failures // TODO: don't ignore failures
@ -3296,7 +3288,7 @@ void ProcessMessage(
// mempool will probably be useless - request the block normally // mempool will probably be useless - request the block normally
std::vector<CInv> vInv(1); std::vector<CInv> vInv(1);
vInv[0] = CInv(MSG_BLOCK | GetFetchFlags(pfrom), cmpctblock.header.GetHash()); vInv[0] = CInv(MSG_BLOCK | GetFetchFlags(pfrom), cmpctblock.header.GetHash());
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETDATA, vInv)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETDATA, vInv));
return; return;
} else { } else {
// If this was an announce-cmpctblock, we want the same treatment as a header message // If this was an announce-cmpctblock, we want the same treatment as a header message
@ -3306,7 +3298,7 @@ void ProcessMessage(
} // cs_main } // cs_main
if (fProcessBLOCKTXN) if (fProcessBLOCKTXN)
return ProcessMessage(pfrom, NetMsgType::BLOCKTXN, blockTxnMsg, time_received, chainparams, chainman, mempool, connman, banman, interruptMsgProc); return ProcessMessage(pfrom, NetMsgType::BLOCKTXN, blockTxnMsg, time_received, chainparams, interruptMsgProc);
if (fRevertToHeaderProcessing) { if (fRevertToHeaderProcessing) {
// Headers received from HB compact block peers are permitted to be // Headers received from HB compact block peers are permitted to be
@ -3314,7 +3306,7 @@ void ProcessMessage(
// the peer if the header turns out to be for an invalid block. // the peer if the header turns out to be for an invalid block.
// Note that if a peer tries to build on an invalid chain, that // Note that if a peer tries to build on an invalid chain, that
// will be detected and the peer will be disconnected/discouraged. // will be detected and the peer will be disconnected/discouraged.
return ProcessHeadersMessage(pfrom, connman, chainman, mempool, {cmpctblock.header}, chainparams, /*via_compact_block=*/true); return ProcessHeadersMessage(pfrom, m_connman, m_chainman, m_mempool, {cmpctblock.header}, chainparams, /*via_compact_block=*/true);
} }
if (fBlockReconstructed) { if (fBlockReconstructed) {
@ -3334,7 +3326,7 @@ void ProcessMessage(
// we have a chain with at least nMinimumChainWork), and we ignore // we have a chain with at least nMinimumChainWork), and we ignore
// compact blocks with less work than our tip, it is safe to treat // compact blocks with less work than our tip, it is safe to treat
// reconstructed compact blocks as having been requested. // reconstructed compact blocks as having been requested.
chainman.ProcessNewBlock(chainparams, pblock, /*fForceProcessing=*/true, &fNewBlock); m_chainman.ProcessNewBlock(chainparams, pblock, /*fForceProcessing=*/true, &fNewBlock);
if (fNewBlock) { if (fNewBlock) {
pfrom.nLastBlockTime = GetTime(); pfrom.nLastBlockTime = GetTime();
} else { } else {
@ -3386,7 +3378,7 @@ void ProcessMessage(
// Might have collided, fall back to getdata now :( // Might have collided, fall back to getdata now :(
std::vector<CInv> invs; std::vector<CInv> invs;
invs.push_back(CInv(MSG_BLOCK | GetFetchFlags(pfrom), resp.blockhash)); invs.push_back(CInv(MSG_BLOCK | GetFetchFlags(pfrom), resp.blockhash));
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETDATA, invs)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::GETDATA, invs));
} else { } else {
// Block is either okay, or possibly we received // Block is either okay, or possibly we received
// READ_STATUS_CHECKBLOCK_FAILED. // READ_STATUS_CHECKBLOCK_FAILED.
@ -3424,7 +3416,7 @@ void ProcessMessage(
// disk-space attacks), but this should be safe due to the // disk-space attacks), but this should be safe due to the
// protections in the compact block handler -- see related comment // protections in the compact block handler -- see related comment
// in compact block optimistic reconstruction handling. // in compact block optimistic reconstruction handling.
chainman.ProcessNewBlock(chainparams, pblock, /*fForceProcessing=*/true, &fNewBlock); m_chainman.ProcessNewBlock(chainparams, pblock, /*fForceProcessing=*/true, &fNewBlock);
if (fNewBlock) { if (fNewBlock) {
pfrom.nLastBlockTime = GetTime(); pfrom.nLastBlockTime = GetTime();
} else { } else {
@ -3458,7 +3450,7 @@ void ProcessMessage(
ReadCompactSize(vRecv); // ignore tx count; assume it is 0. ReadCompactSize(vRecv); // ignore tx count; assume it is 0.
} }
return ProcessHeadersMessage(pfrom, connman, chainman, mempool, headers, chainparams, /*via_compact_block=*/false); return ProcessHeadersMessage(pfrom, m_connman, m_chainman, m_mempool, headers, chainparams, /*via_compact_block=*/false);
} }
if (msg_type == NetMsgType::BLOCK) if (msg_type == NetMsgType::BLOCK)
@ -3487,7 +3479,7 @@ void ProcessMessage(
mapBlockSource.emplace(hash, std::make_pair(pfrom.GetId(), true)); mapBlockSource.emplace(hash, std::make_pair(pfrom.GetId(), true));
} }
bool fNewBlock = false; bool fNewBlock = false;
chainman.ProcessNewBlock(chainparams, pblock, forceProcessing, &fNewBlock); m_chainman.ProcessNewBlock(chainparams, pblock, forceProcessing, &fNewBlock);
if (fNewBlock) { if (fNewBlock) {
pfrom.nLastBlockTime = GetTime(); pfrom.nLastBlockTime = GetTime();
} else { } else {
@ -3523,9 +3515,9 @@ void ProcessMessage(
pfrom.vAddrToSend.clear(); pfrom.vAddrToSend.clear();
std::vector<CAddress> vAddr; std::vector<CAddress> vAddr;
if (pfrom.HasPermission(PF_ADDR)) { if (pfrom.HasPermission(PF_ADDR)) {
vAddr = connman.GetAddresses(MAX_ADDR_TO_SEND, MAX_PCT_ADDR_TO_SEND); vAddr = m_connman.GetAddresses(MAX_ADDR_TO_SEND, MAX_PCT_ADDR_TO_SEND);
} else { } else {
vAddr = connman.GetAddresses(pfrom.addr.GetNetwork(), MAX_ADDR_TO_SEND, MAX_PCT_ADDR_TO_SEND); vAddr = m_connman.GetAddresses(pfrom.addr.GetNetwork(), MAX_ADDR_TO_SEND, MAX_PCT_ADDR_TO_SEND);
} }
FastRandomContext insecure_rand; FastRandomContext insecure_rand;
for (const CAddress &addr : vAddr) { for (const CAddress &addr : vAddr) {
@ -3545,7 +3537,7 @@ void ProcessMessage(
return; return;
} }
if (connman.OutboundTargetReached(false) && !pfrom.HasPermission(PF_MEMPOOL)) if (m_connman.OutboundTargetReached(false) && !pfrom.HasPermission(PF_MEMPOOL))
{ {
if (!pfrom.HasPermission(PF_NOBAN)) if (!pfrom.HasPermission(PF_NOBAN))
{ {
@ -3578,7 +3570,7 @@ void ProcessMessage(
// it, if the remote node sends a ping once per second and this node takes 5 // it, if the remote node sends a ping once per second and this node takes 5
// seconds to respond to each, the 5th ping the remote sends would appear to // seconds to respond to each, the 5th ping the remote sends would appear to
// return very quickly. // return very quickly.
connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::PONG, nonce)); m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::PONG, nonce));
} }
return; return;
} }
@ -3718,17 +3710,17 @@ void ProcessMessage(
} }
if (msg_type == NetMsgType::GETCFILTERS) { if (msg_type == NetMsgType::GETCFILTERS) {
ProcessGetCFilters(pfrom, vRecv, chainparams, connman); ProcessGetCFilters(pfrom, vRecv, chainparams, m_connman);
return; return;
} }
if (msg_type == NetMsgType::GETCFHEADERS) { if (msg_type == NetMsgType::GETCFHEADERS) {
ProcessGetCFHeaders(pfrom, vRecv, chainparams, connman); ProcessGetCFHeaders(pfrom, vRecv, chainparams, m_connman);
return; return;
} }
if (msg_type == NetMsgType::GETCFCHECKPT) { if (msg_type == NetMsgType::GETCFCHECKPT) {
ProcessGetCFCheckPt(pfrom, vRecv, chainparams, connman); ProcessGetCFCheckPt(pfrom, vRecv, chainparams, m_connman);
return; return;
} }
@ -3886,7 +3878,7 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic<bool>& inter
} }
try { try {
ProcessMessage(*pfrom, msg_type, vRecv, msg.m_time, chainparams, m_chainman, m_mempool, m_connman, m_banman, interruptMsgProc); ProcessMessage(*pfrom, msg_type, vRecv, msg.m_time, chainparams, interruptMsgProc);
if (interruptMsgProc) if (interruptMsgProc)
return false; return false;
if (!pfrom->vRecvGetData.empty()) if (!pfrom->vRecvGetData.empty())

7
src/net_processing.h
View file

@ -11,6 +11,7 @@
#include <sync.h> #include <sync.h>
#include <validationinterface.h> #include <validationinterface.h>
class CChainParams;
class CTxMemPool; class CTxMemPool;
class ChainstateManager; class ChainstateManager;
@ -85,8 +86,14 @@ public:
/** Retrieve unbroadcast transactions from the mempool and reattempt sending to peers */ /** Retrieve unbroadcast transactions from the mempool and reattempt sending to peers */
void ReattemptInitialBroadcast(CScheduler& scheduler) const; void ReattemptInitialBroadcast(CScheduler& scheduler) const;
/** Process a single message from a peer. Public for fuzz testing */
void ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv,
const std::chrono::microseconds time_received, const CChainParams& chainparams,
const std::atomic<bool>& interruptMsgProc);
private: private:
int64_t m_stale_tip_check_time; //!< Next time to check for stale tip int64_t m_stale_tip_check_time; //!< Next time to check for stale tip
}; };
struct CNodeStateStats { struct CNodeStateStats {

17
src/test/fuzz/process_message.cpp
View file

@ -30,18 +30,6 @@
#include <string> #include <string>
#include <vector> #include <vector>
void ProcessMessage(
CNode& pfrom,
const std::string& msg_type,
CDataStream& vRecv,
const std::chrono::microseconds time_received,
const CChainParams& chainparams,
ChainstateManager& chainman,
CTxMemPool& mempool,
CConnman& connman,
BanMan* banman,
const std::atomic<bool>& interruptMsgProc);
namespace { namespace {
#ifdef MESSAGE_TYPE #ifdef MESSAGE_TYPE
@ -87,9 +75,8 @@ void test_one_input(const std::vector<uint8_t>& buffer)
connman.AddTestNode(p2p_node); connman.AddTestNode(p2p_node);
g_setup->m_node.peer_logic->InitializeNode(&p2p_node); g_setup->m_node.peer_logic->InitializeNode(&p2p_node);
try { try {
ProcessMessage(p2p_node, random_message_type, random_bytes_data_stream, GetTime<std::chrono::microseconds>(), g_setup->m_node.peer_logic->ProcessMessage(p2p_node, random_message_type, random_bytes_data_stream,
Params(), *g_setup->m_node.chainman, *g_setup->m_node.mempool, GetTime<std::chrono::microseconds>(), Params(),
*g_setup->m_node.connman, g_setup->m_node.banman.get(),
std::atomic<bool>{false}); std::atomic<bool>{false});
} catch (const std::ios_base::failure&) { } catch (const std::ios_base::failure&) {
} }