From 87d281029e1b157afc9cfbb8d20a5373968c84c4 Mon Sep 17 00:00:00 2001 From: j-berman Date: Fri, 7 Aug 2026 12:21:00 -0700 Subject: [PATCH 1/2] cryptonote_basic: parse_and_validate_tx param to check max size Simple API to check max blob size before parsing. This is useful when reading blobs from untrusted sources. @selsta pointed out that coinbase txs can technically be larger than get_max_tx_size(), otherwise we could enforce it on all txs. --- .../cryptonote_format_utils.cpp | 23 ++++++++++++++----- .../cryptonote_format_utils.h | 11 +++++---- src/cryptonote_core/blockchain.cpp | 2 +- src/cryptonote_core/cryptonote_core.cpp | 5 ++-- src/cryptonote_core/tx_sanity_check.cpp | 2 +- .../cryptonote_protocol_handler.inl | 13 ++++------- src/wallet/wallet2.cpp | 2 +- 7 files changed, 33 insertions(+), 25 deletions(-) diff --git a/src/cryptonote_basic/cryptonote_format_utils.cpp b/src/cryptonote_basic/cryptonote_format_utils.cpp index a0e04dca140..1af074edbbe 100644 --- a/src/cryptonote_basic/cryptonote_format_utils.cpp +++ b/src/cryptonote_basic/cryptonote_format_utils.cpp @@ -156,6 +156,13 @@ namespace cryptonote return tx.vout.size(); } //--------------------------------------------------------------- + bool passes_max_size_check(const bool max_size_check, const blobdata_ref &tx_blob) + { + if (!max_size_check) + return true; + return tx_blob.size() <= get_max_tx_size(); + } + //--------------------------------------------------------------- } namespace cryptonote @@ -254,8 +261,9 @@ namespace cryptonote return true; } //--------------------------------------------------------------- - bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx) + bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, const bool max_size_check) { + CHECK_AND_ASSERT_MES(passes_max_size_check(max_size_check, tx_blob), false, "Tx blob too big"); binary_archive ba{epee::strspan(tx_blob)}; bool r = ::serialization::serialize(ba, tx); CHECK_AND_ASSERT_MES(r, false, "Failed to parse transaction from blob"); @@ -265,8 +273,9 @@ namespace cryptonote return true; } //--------------------------------------------------------------- - bool parse_and_validate_tx_base_from_blob(const blobdata_ref& tx_blob, transaction& tx) + bool parse_and_validate_tx_base_from_blob(const blobdata_ref& tx_blob, transaction& tx, const bool max_size_check) { + CHECK_AND_ASSERT_MES(passes_max_size_check(max_size_check, tx_blob), false, "Tx blob too big"); binary_archive ba{epee::strspan(tx_blob)}; bool r = tx.serialize_base(ba); CHECK_AND_ASSERT_MES(r, false, "Failed to parse transaction from blob"); @@ -275,16 +284,18 @@ namespace cryptonote return true; } //--------------------------------------------------------------- - bool parse_and_validate_tx_prefix_from_blob(const blobdata_ref& tx_blob, transaction_prefix& tx) + bool parse_and_validate_tx_prefix_from_blob(const blobdata_ref& tx_blob, transaction_prefix& tx, const bool max_size_check) { + CHECK_AND_ASSERT_MES(passes_max_size_check(max_size_check, tx_blob), false, "Tx blob too big"); binary_archive ba{epee::strspan(tx_blob)}; bool r = ::serialization::serialize_noeof(ba, tx); CHECK_AND_ASSERT_MES(r, false, "Failed to parse transaction prefix from blob"); return true; } //--------------------------------------------------------------- - bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, crypto::hash& tx_hash) + bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, crypto::hash& tx_hash, const bool max_size_check) { + CHECK_AND_ASSERT_MES(passes_max_size_check(max_size_check, tx_blob), false, "Tx blob too big"); binary_archive ba{epee::strspan(tx_blob)}; bool r = ::serialization::serialize(ba, tx); CHECK_AND_ASSERT_MES(r, false, "Failed to parse transaction from blob"); @@ -296,9 +307,9 @@ namespace cryptonote return get_transaction_hash(tx, tx_hash); } //--------------------------------------------------------------- - bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, crypto::hash& tx_hash, crypto::hash& tx_prefix_hash) + bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, crypto::hash& tx_hash, crypto::hash& tx_prefix_hash, const bool max_size_check) { - if (!parse_and_validate_tx_from_blob(tx_blob, tx, tx_hash)) + if (!parse_and_validate_tx_from_blob(tx_blob, tx, tx_hash, max_size_check)) return false; get_transaction_prefix_hash(tx, tx_prefix_hash); return true; diff --git a/src/cryptonote_basic/cryptonote_format_utils.h b/src/cryptonote_basic/cryptonote_format_utils.h index b4956bb54a3..8927b59f4b5 100644 --- a/src/cryptonote_basic/cryptonote_format_utils.h +++ b/src/cryptonote_basic/cryptonote_format_utils.h @@ -55,12 +55,13 @@ namespace cryptonote crypto::hash get_transaction_prefix_hash(const transaction_prefix& tx, hw::device &hwdev); void get_transaction_prefix_hash(const transaction_prefix& tx, crypto::hash& h); crypto::hash get_transaction_prefix_hash(const transaction_prefix& tx); - bool parse_and_validate_tx_prefix_from_blob(const blobdata_ref& tx_blob, transaction_prefix& tx); bool expand_transaction_1(transaction &tx, bool base_only); - bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, crypto::hash& tx_hash, crypto::hash& tx_prefix_hash); - bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, crypto::hash& tx_hash); - bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx); - bool parse_and_validate_tx_base_from_blob(const blobdata_ref& tx_blob, transaction& tx); + /* The size check is useful before parsing non-coinbase txs from untrusted sources. Coinbase blob sizes may be uncapped. */ + bool parse_and_validate_tx_prefix_from_blob(const blobdata_ref& tx_blob, transaction_prefix& tx, const bool max_size_check = false); + bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, crypto::hash& tx_hash, crypto::hash& tx_prefix_hash, const bool max_size_check = false); + bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, crypto::hash& tx_hash, const bool max_size_check = false); + bool parse_and_validate_tx_from_blob(const blobdata_ref& tx_blob, transaction& tx, const bool max_size_check = false); + bool parse_and_validate_tx_base_from_blob(const blobdata_ref& tx_blob, transaction& tx, const bool max_size_check = false); bool is_v1_tx(const blobdata_ref& tx_blob); bool is_v1_tx(const blobdata& tx_blob); diff --git a/src/cryptonote_core/blockchain.cpp b/src/cryptonote_core/blockchain.cpp index 6e96bf04283..a25039271d7 100644 --- a/src/cryptonote_core/blockchain.cpp +++ b/src/cryptonote_core/blockchain.cpp @@ -5689,7 +5689,7 @@ bool Blockchain::prepare_handle_incoming_blocks(const std::vector get_max_tx_size()) - { - MERROR("Transaction blob of length " << tx_entry.blob.size() << " is too large to unpack!"); - return false; - } - const bool is_pruned = tx_entry.prunable_hash != crypto::null_hash; if (is_pruned && !allow_pruned) { @@ -120,14 +114,15 @@ namespace cryptonote cryptonote::transaction tx; crypto::hash tx_hash; bool parse_success = false; + const bool max_size_check = true; if (is_pruned) { - if ((parse_success = cryptonote::parse_and_validate_tx_base_from_blob(tx_entry.blob, tx))) + if ((parse_success = cryptonote::parse_and_validate_tx_base_from_blob(tx_entry.blob, tx, max_size_check))) parse_success = cryptonote::get_pruned_transaction_hash(tx, tx_entry.prunable_hash, tx_hash); } else { - parse_success = cryptonote::parse_and_validate_tx_from_blob(tx_entry.blob, tx, tx_hash); + parse_success = cryptonote::parse_and_validate_tx_from_blob(tx_entry.blob, tx, tx_hash, max_size_check); } if (!parse_success) @@ -1030,7 +1025,7 @@ namespace cryptonote std::lock_guard m_check_lock(m_check_tx_request_queue_mutex); for (const auto &blob: arg.txs) { - MLOGIF_P2P_MESSAGE(cryptonote::transaction tx; crypto::hash hash; bool ret = cryptonote::parse_and_validate_tx_from_blob(blob, tx, hash);, ret, "Including transaction " << hash); + MLOGIF_P2P_MESSAGE(cryptonote::transaction tx; crypto::hash hash; bool ret = cryptonote::parse_and_validate_tx_from_blob(blob, tx, hash, true);, ret, "Including transaction " << hash); if (seen.find(blob) != seen.end()) { LOG_PRINT_CCONTEXT_L1("Duplicate transaction in notification, dropping connection"); diff --git a/src/wallet/wallet2.cpp b/src/wallet/wallet2.cpp index e818afe530b..05cb4b7a518 100644 --- a/src/wallet/wallet2.cpp +++ b/src/wallet/wallet2.cpp @@ -3077,7 +3077,7 @@ void wallet2::process_pool_info_extent(const cryptonote::COMMAND_RPC_GET_BLOCKS_ for (const auto &pool_tx: res.added_pool_txs) { cryptonote::transaction tx; - THROW_WALLET_EXCEPTION_IF(!cryptonote::parse_and_validate_tx_base_from_blob(pool_tx.tx_blob, tx), + THROW_WALLET_EXCEPTION_IF(!cryptonote::parse_and_validate_tx_base_from_blob(pool_tx.tx_blob, tx, true), error::wallet_internal_error, "Failed to validate transaction base from daemon"); added_pool_txs.emplace_back(std::move(tx), pool_tx.tx_hash, pool_tx.double_spend_seen); } From 777ed68a2b6a7abb2205d2df5a505c46284483ac Mon Sep 17 00:00:00 2001 From: j-berman Date: Fri, 17 Jul 2026 10:17:32 -0700 Subject: [PATCH 2/2] tx relay v2: more improvements 1. Remove unnecessary locking from the functions that strictly read the db for pool txs (`tx_memory_pool::get_transaction` and `tx_memory_pool::have_tx`). 2. Don't ban peers if they miss too many tx requests (drop connection, but don't ban). 3. Implement a nonce in the p2p tx hash / tx notify messages, so that we can track exact request -> response. 4. Remove the lock synchronizing `handle_notify_new_transactions` and the 5. Request txs from peers as soon as our local capacity to accept more txs 6. Fix bugged logic in `handle_notify_tx_pool_hash` that starts the timer 7. Fix bugged logic in `handle_notify_tx_pool_hash` that could result in us attempting to request **more** than the allowed max, and thus not requesting some tx hashes correctly. 8. Use boost multi-index's `.modify()` to update elems in the `request_manager`, rather than updating the iterator in place. 9. Align fluff timer flush_time for tx relay v2. --- .../blockchain_import.cpp | 12 +- src/cryptonote_core/cryptonote_core.cpp | 11 +- src/cryptonote_core/cryptonote_core.h | 7 +- src/cryptonote_core/tx_pool.cpp | 12 +- .../cryptonote_protocol_defs.h | 4 + .../cryptonote_protocol_handler.h | 6 +- .../cryptonote_protocol_handler.inl | 136 +++++++++------ src/cryptonote_protocol/levin_notify.cpp | 2 +- src/cryptonote_protocol/request_manager.cpp | 157 ++++++++++++++---- src/cryptonote_protocol/request_manager.h | 34 +++- src/cryptonote_protocol/txrequestqueue.h | 36 +++- src/rpc/core_rpc_server.cpp | 20 +-- src/rpc/daemon_handler.cpp | 5 +- tests/core_tests/chaingen.h | 32 ++-- tests/fuzz/fuzz_rpc/initialisation.cpp | 18 +- tests/unit_tests/node_server.cpp | 2 +- tests/unit_tests/request_manager.cpp | 104 ++++++++++-- 17 files changed, 428 insertions(+), 170 deletions(-) diff --git a/src/blockchain_utilities/blockchain_import.cpp b/src/blockchain_utilities/blockchain_import.cpp index b00e407b56b..5dae1db0642 100644 --- a/src/blockchain_utilities/blockchain_import.cpp +++ b/src/blockchain_utilities/blockchain_import.cpp @@ -175,13 +175,15 @@ int check_flush(cryptonote::core &core, std::vector &block tx_verification_context tvc = AUTO_VAL_INIT(tvc); CHECK_AND_ASSERT_THROW_MES(tx_blob.prunable_hash == crypto::null_hash, "block entry must not contain pruned txs"); + cryptonote::transaction tx; crypto::hash txid; - core.handle_incoming_tx(tx_blob.blob, tvc, relay_method::block, true, txid); - if(tvc.m_verifivation_failed) + const bool parse_success = cryptonote::parse_and_validate_tx_from_blob(tx_blob.blob, tx, txid, true); + if (!parse_success + || !core.handle_incoming_tx(tx_blob.blob, tx, txid, tvc, relay_method::block, true) + || tvc.m_verifivation_failed) { - cryptonote::transaction transaction; - if (cryptonote::parse_and_validate_tx_from_blob(tx_blob.blob, transaction)) - MERROR("Transaction verification failed, tx_id = " << cryptonote::get_transaction_hash(transaction)); + if (parse_success) + MERROR("Transaction verification failed, tx_id = " << txid); else MERROR("Transaction verification failed, transaction is unparsable"); core.cleanup_handle_incoming_blocks(); diff --git a/src/cryptonote_core/cryptonote_core.cpp b/src/cryptonote_core/cryptonote_core.cpp index a05fa50bd3f..0604134847b 100644 --- a/src/cryptonote_core/cryptonote_core.cpp +++ b/src/cryptonote_core/cryptonote_core.cpp @@ -781,7 +781,7 @@ namespace cryptonote return true; } //----------------------------------------------------------------------------------------------- - bool core::handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, relay_method tx_relay, bool relayed, crypto::hash& txid) + bool core::handle_incoming_tx(const blobdata& tx_blob, transaction& tx, const crypto::hash& txid, tx_verification_context& tvc, relay_method tx_relay, bool relayed) { tvc = {}; @@ -797,15 +797,6 @@ namespace cryptonote return false; } - transaction tx; - crypto::hash txid; - if (!parse_and_validate_tx_from_blob(tx_blob, tx, txid, true)) - { - LOG_PRINT_L1("Incoming transactions failed to parse, rejected"); - tvc.m_verifivation_failed = true; - return false; - } - const uint64_t tx_weight = get_transaction_weight(tx, tx_blob.size()); if (!add_new_tx(tx, txid, tx_blob, tx_weight, tvc, tx_relay, relayed)) return false; diff --git a/src/cryptonote_core/cryptonote_core.h b/src/cryptonote_core/cryptonote_core.h index 28ad59bef8b..367f24f3b3c 100644 --- a/src/cryptonote_core/cryptonote_core.h +++ b/src/cryptonote_core/cryptonote_core.h @@ -116,18 +116,19 @@ namespace cryptonote /** * @brief handles an incoming transaction * - * Parses an incoming transaction and, if nothing is obviously wrong, + * Processes an incoming transaction and, if nothing is obviously wrong, * passes it along to the transaction pool * * @param tx_blob the tx to handle + * @param tx the parsed tx to handle (may expand the tx) + * @param txid the tx hash to handle * @param tvc metadata about the transaction's validity * @param tx_relay how the transaction was received * @param relayed whether or not the transaction was relayed to us - * @param txid return by reference * * @return true if the transaction was accepted, false otherwise */ - bool handle_incoming_tx(const blobdata& tx_blob, tx_verification_context& tvc, relay_method tx_relay, bool relayed, crypto::hash& txid); + bool handle_incoming_tx(const blobdata& tx_blob, transaction& tx, const crypto::hash& txid, tx_verification_context& tvc, relay_method tx_relay, bool relayed); /** * @brief handles a single incoming block diff --git a/src/cryptonote_core/tx_pool.cpp b/src/cryptonote_core/tx_pool.cpp index 195aed27afe..0538cc03f52 100644 --- a/src/cryptonote_core/tx_pool.cpp +++ b/src/cryptonote_core/tx_pool.cpp @@ -1422,8 +1422,10 @@ namespace cryptonote //--------------------------------------------------------------------------------- bool tx_memory_pool::get_transaction(const crypto::hash& id, cryptonote::blobdata& txblob, relay_category tx_category) const { - CRITICAL_REGION_LOCAL(m_transactions_lock); - CRITICAL_REGION_LOCAL1(m_blockchain); + // WARNING: this function does not take m_blockchain_lock, and thus should only call read only + // m_db functions which do not depend on one another (ie, no getheight + gethash(height-1), as + // well as not accessing class members, even read only (ie, m_invalid_blocks). The caller must + // lock if it is otherwise needed. try { return m_blockchain.get_txpool_tx_blob(id, txblob, tx_category); @@ -1452,8 +1454,10 @@ namespace cryptonote //--------------------------------------------------------------------------------- bool tx_memory_pool::have_tx(const crypto::hash &id, relay_category tx_category) const { - CRITICAL_REGION_LOCAL(m_transactions_lock); - CRITICAL_REGION_LOCAL1(m_blockchain); + // WARNING: this function does not take m_blockchain_lock, and thus should only call read only + // m_db functions which do not depend on one another (ie, no getheight + gethash(height-1), as + // well as not accessing class members, even read only (ie, m_invalid_blocks). The caller must + // lock if it is otherwise needed. return m_blockchain.get_db().txpool_has_tx(id, tx_category); } //--------------------------------------------------------------------------------- diff --git a/src/cryptonote_protocol/cryptonote_protocol_defs.h b/src/cryptonote_protocol/cryptonote_protocol_defs.h index 4907f0476cf..07aa544186c 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_defs.h +++ b/src/cryptonote_protocol/cryptonote_protocol_defs.h @@ -200,11 +200,13 @@ namespace cryptonote std::vector txs; std::string _; // padding bool dandelionpp_fluff; //zero initialization defaults to stem mode + uint64_t nonce; // if responding to NOTIFY_REQUEST_TX_POOL_TXS, includes nonce in resp BEGIN_KV_SERIALIZE_MAP() KV_SERIALIZE(txs) KV_SERIALIZE(_) KV_SERIALIZE_OPT(dandelionpp_fluff, true) // backwards compatible mode is fluff + KV_SERIALIZE_OPT(nonce, (uint64_t)0) END_KV_SERIALIZE_MAP() }; typedef epee::misc_utils::struct_init request; @@ -445,9 +447,11 @@ namespace cryptonote struct request_t { + uint64_t n; // request nonce std::vector t; BEGIN_KV_SERIALIZE_MAP() + KV_SERIALIZE(n) KV_SERIALIZE_CONTAINER_POD_AS_BLOB(t) END_KV_SERIALIZE_MAP() }; diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.h b/src/cryptonote_protocol/cryptonote_protocol_handler.h index 4f6ccfd4bdd..833700b5676 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler.h +++ b/src/cryptonote_protocol/cryptonote_protocol_handler.h @@ -176,14 +176,14 @@ namespace cryptonote bool should_ask_for_pruned_data(cryptonote_connection_context& context, uint64_t first_block_height, uint64_t nblocks, bool check_block_weights) const; void drop_connection(cryptonote_connection_context &context, bool add_fail, bool flush_all_spans); void drop_connection_with_score(cryptonote_connection_context &context, unsigned int score, bool flush_all_spans); - void drop_connection(const boost::uuids::uuid&); + void drop_connection(const boost::uuids::uuid&, bool add_fail); void drop_connections(const epee::net_utils::network_address address); bool kick_idle_peers(); bool check_standby_peers(); bool update_sync_search(); - void send_txs_request(cryptonote_connection_context &context, std::vector &&tx_hashes); - std::mutex m_check_tx_request_queue_mutex; + void send_txs_request(cryptonote_connection_context &context, request_manager::tx_request_t &&tx_req); bool check_tx_request_queue(); + void fly_available_requests_in_queue(const std::unordered_set &ignore_peers = {}); int try_add_next_blocks(cryptonote_connection_context &context); void notify_new_stripe(cryptonote_connection_context &context, uint32_t stripe); size_t skip_unneeded_hashes(cryptonote_connection_context& context, bool check_block_queue) const; diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.inl b/src/cryptonote_protocol/cryptonote_protocol_handler.inl index 0502bd0af2f..fa2e03bd0a3 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler.inl +++ b/src/cryptonote_protocol/cryptonote_protocol_handler.inl @@ -775,6 +775,7 @@ namespace cryptonote relay_block(arg, context); for (const auto &tx_hash : new_block.tx_hashes) m_request_manager.remove_request(tx_hash); + this->fly_available_requests_in_queue(); } else if( bvc.m_marked_as_orphaned ) { @@ -956,12 +957,13 @@ namespace cryptonote m_request_manager.remove_request(tx_hash); continue; } - const bool send_request = m_request_manager.add_request(tx_hash, context.m_connection_id); - if (send_request) - missing_tx_hashes.push_back(tx_hash); + missing_tx_hashes.push_back(tx_hash); } - this->send_txs_request(context, std::move(missing_tx_hashes)); + // Add all the missing to our request queue, and then kick off the request + auto txs_req = m_request_manager.enqueue_requests(missing_tx_hashes, context.m_connection_id); + this->send_txs_request(context, std::move(txs_req)); + this->fly_available_requests_in_queue(); return 1; } @@ -988,18 +990,22 @@ namespace cryptonote { txs.push_back(std::move(tx_blob)); } - // If tx is not in the pool, then ignore it (do not penalize peer) + else + { + // If tx is not in the pool, then ignore it (do not penalize peer) + MLOG_P2P_MESSAGE("Requested tx " << tx_hash << " not found in pool"); + } } - // Send response if any txs found - if (!txs.empty()) - { - NOTIFY_NEW_TRANSACTIONS::request request = {}; - request.txs = std::move(txs); - request.dandelionpp_fluff = true; - pad_tx_request(request); - post_notify(request, context); - } + MLOG_P2P_MESSAGE("Sending " << txs.size() << " back to peer (nonce=" << arg.n << ")"); + + // Send response including the nonce that was requested, even if no txs included in resp + NOTIFY_NEW_TRANSACTIONS::request request = {}; + request.txs = std::move(txs); + request.dandelionpp_fluff = true; + request.nonce = arg.n; + pad_tx_request(request); + post_notify(request, context); return 1; } @@ -1021,20 +1027,44 @@ namespace cryptonote return 1; } - std::unordered_set seen; - std::lock_guard m_check_lock(m_check_tx_request_queue_mutex); - for (const auto &blob: arg.txs) + // Parse the txs and prevent duplicates + std::vector parsed_txs; + std::vector tx_hashes; + parsed_txs.reserve(arg.txs.size()); + tx_hashes.reserve(arg.txs.size()); { - MLOGIF_P2P_MESSAGE(cryptonote::transaction tx; crypto::hash hash; bool ret = cryptonote::parse_and_validate_tx_from_blob(blob, tx, hash, true);, ret, "Including transaction " << hash); - if (seen.find(blob) != seen.end()) + std::unordered_set seen; + bool already_seen = false; + bool parse_failed = false; + for (auto& tx_blob : arg.txs) + { + already_seen = seen.find(tx_blob) != seen.end(); + if (already_seen) + break; + parse_failed = !cryptonote::parse_and_validate_tx_from_blob(tx_blob, parsed_txs.emplace_back(), tx_hashes.emplace_back(), true); + if (parse_failed) + break; + seen.insert(tx_blob); + MLOG_P2P_MESSAGE("Including tx " << tx_hashes.back()); + } + + if (already_seen || parse_failed) { - LOG_PRINT_CCONTEXT_L1("Duplicate transaction in notification, dropping connection"); + if (already_seen) + LOG_PRINT_CCONTEXT_L1("Duplicate transaction in notification, dropping connection"); + else if (parse_failed) + LOG_PRINT_CCONTEXT_L1("Failed to parse incoming tx, dropping connection"); drop_connection(context, false, false); return 1; } - seen.insert(blob); } + /* Indicate we're processing the txs so that in case processing takes a + while, we won't think any requests for the txs from that peer are stale. + We only mark a tx as processing for a specific peer, because we may still + be expecting some other peer to send us the tx. */ + m_request_manager.processing_txs(context.m_connection_id, arg.nonce, tx_hashes); + /* If the txes were received over i2p/tor, the default is to "forward" with a randomized delay to further enhance the "white noise" behavior, potentially making it harder for ISP-level spies to determine which @@ -1060,19 +1090,21 @@ namespace cryptonote else stem_txs.reserve(arg.txs.size()); - for (auto& tx_blob : arg.txs) + for (std::size_t i = 0; i < arg.txs.size(); ++i) { + auto &tx_blob = arg.txs.at(i); + auto tx = std::move(parsed_txs.at(i)); + auto tx_hash = std::move(tx_hashes.at(i)); + tx_verification_context tvc{}; - crypto::hash tx_hash{}; - if (!m_core.handle_incoming_tx(tx_blob, tvc, tx_relay, true, tx_hash) && !tvc.m_no_drop_offense) + if (!m_core.handle_incoming_tx(tx_blob, tx, tx_hash, tvc, tx_relay, true) && !tvc.m_no_drop_offense) { LOG_PRINT_CCONTEXT_L1("Tx verification failed, dropping connection"); drop_connection(context, false, false); return 1; } - if (tx_hash != crypto::hash{}) - m_request_manager.remove_request(tx_hash); + m_request_manager.remove_request(tx_hash); switch (tvc.m_relay) { @@ -1107,6 +1139,8 @@ namespace cryptonote arg.txs = std::move(fluff_txs); relay_transactions(arg, std::move(fluff_hashes), context.m_connection_id, context.m_remote_address.get_zone(), relay_method::fluff); } + + this->fly_available_requests_in_queue(); return 1; } //------------------------------------------------------------------------------------------------------------------------ @@ -1533,7 +1567,7 @@ namespace cryptonote if (confirmed_height != std::numeric_limits::max() && confirmed_height + 1 != start_height) { MERROR(context << "Found incorrect height for " << new_block.prev_id << " provided by " << span_connection_id); - drop_connection(span_connection_id); + drop_connection(span_connection_id, true); return 1; } @@ -1702,6 +1736,7 @@ namespace cryptonote { for (const auto &h : *tx_hashes_ptr) m_request_manager.remove_request(h); + this->fly_available_requests_in_queue(); } } @@ -1902,29 +1937,25 @@ skip: } //------------------------------------------------------------------------------------------------------------------------ template - void t_cryptonote_protocol_handler::send_txs_request(cryptonote_connection_context& context, std::vector &&tx_hashes) + void t_cryptonote_protocol_handler::send_txs_request(cryptonote_connection_context& context, request_manager::tx_request_t &&tx_req) { + auto &&tx_hashes = std::move(tx_req.tx_hashes); if (tx_hashes.empty()) - { - MLOG_P2P_MESSAGE("Not sending a request for txs"); return; - } // We *never* expect to call this with more txs than can be sent in a packet CHECK_AND_ASSERT_MES(tx_hashes.size() <= max_n_txs_per_packet(),, "Too many txs in NOTIFY_REQUEST_TX_POOL_TXS"); NOTIFY_REQUEST_TX_POOL_TXS::request req; + req.n = tx_req.nonce; req.t = std::move(tx_hashes); - MLOG_P2P_MESSAGE("Requesting " << req.t.size() << " transactions via NOTIFY_REQUEST_TX_POOL_TXS"); + MLOG_P2P_MESSAGE("Requesting " << req.t.size() << " transactions via NOTIFY_REQUEST_TX_POOL_TXS (nonce=" << req.n << ")"); post_notify(req, context); } //----------------------------------------------------------------------------------------------------------------------- template bool t_cryptonote_protocol_handler::check_tx_request_queue() { - // We want to check this frequently, so we keep making sure tx requests that aren't in flight get placed in flight. - // At time of writing, I set this to run every 5s, and the default timeout for stale requests is 30s. - // If we're not synchronized, we shouldn't be requesting any txs. Syncing might end up removing many tx requests // because the txs enter the chain. if (!is_synchronized()) @@ -1935,25 +1966,28 @@ skip: MCTRACE("net.p2p.msg", "on_idle :: check_tx_request_queue, starting ..."); - // Synchronize with handling incoming txs, because that function can take a long time to execute and may be in - // the process of verifying large txs that we requested. We don't want to count request misses that are actually - // good and just take a long time to verify. - std::lock_guard m_check_lock(m_check_tx_request_queue_mutex); - // We drop connections that exceed the threshold for allowed missed txs const auto drop_peers = m_request_manager.remove_stale_requests(); for (const auto &peer_id : drop_peers) { MCINFO("net.p2p.msg", "Missed tx request more than threshold of the time, dropping peer : " << epee::string_tools::pod_to_hex(peer_id)); - drop_connection(peer_id); + // Don't want to ban peers at least for now, since we've observed honest peers get banned due to long response + // times on stressnet. We can revisit this decision if we observe close to zero honest dropped conns under stress. + drop_connection(peer_id, false); } - // Let fly any queued tx requests that our connections can handle. This is the section that benefits from calling - // check_tx_request_queue more frequently than the timeout. Since connections can become able to handle new - // requests frequently as we process incoming txs. + this->fly_available_requests_in_queue(drop_peers); + return true; + } + //----------------------------------------------------------------------------------------------------------------------- + template + void t_cryptonote_protocol_handler::fly_available_requests_in_queue(const std::unordered_set &ignore_peers) + { + // Let fly any queued tx requests that our connections can handle. This is good to call after removing requests from + // the request manager, since a connection might have become available to handle more requests. m_p2p->for_each_connection([&](cryptonote_connection_context& context, nodetool::peerid_type _unused, uint32_t _unused2)->bool { - if (drop_peers.count(context.m_connection_id)) + if (ignore_peers.count(context.m_connection_id)) { MDEBUG(context << "connection is set to be dropped, not sending more tx requests"); return true; @@ -1966,13 +2000,11 @@ skip: return true; } - std::vector new_requests = m_request_manager.fly_available_requests(context.m_connection_id); - this->send_txs_request(context, std::move(new_requests)); + auto tx_req = m_request_manager.fly_available_requests(context.m_connection_id); + this->send_txs_request(context, std::move(tx_req)); return true; }); - - return true; } //------------------------------------------------------------------------------------------------------------------------ template @@ -3033,11 +3065,11 @@ skip: } //------------------------------------------------------------------------------------------------------------------------ template - void t_cryptonote_protocol_handler::drop_connection(const boost::uuids::uuid& id) + void t_cryptonote_protocol_handler::drop_connection(const boost::uuids::uuid& id, bool add_fail) { - m_p2p->for_connection(id, [this](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{ + m_p2p->for_connection(id, [this, add_fail](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{ // This _could be_ outside of strand, so careful on actions - drop_connection(context, true, false); + drop_connection(context, add_fail, false); return true; }); } diff --git a/src/cryptonote_protocol/levin_notify.cpp b/src/cryptonote_protocol/levin_notify.cpp index 4f0cd77a92f..14f16792f57 100644 --- a/src/cryptonote_protocol/levin_notify.cpp +++ b/src/cryptonote_protocol/levin_notify.cpp @@ -463,7 +463,7 @@ namespace levin // When i2p/tor, only fluff to outbound connections if (source != id && (zone->nzone == epee::net_utils::zone::public_ || !context.m_is_income)) { - if (context.fluff_txs.empty()) + if (context.tx_relay_v2 ? context.fluff_txs_v2.empty() : context.fluff_txs.empty()) context.flush_time = now + (context.m_is_income ? in_duration() : out_duration()); next_flush = std::min(next_flush, context.flush_time); diff --git a/src/cryptonote_protocol/request_manager.cpp b/src/cryptonote_protocol/request_manager.cpp index a6c1873178a..83a28a4e5ae 100644 --- a/src/cryptonote_protocol/request_manager.cpp +++ b/src/cryptonote_protocol/request_manager.cpp @@ -38,6 +38,38 @@ #include #include +/* + Using boost multi-index's modify() to update elems ensures all indexes + are updated in addition to the underlying element, without doing any copies. + modify() fails when there is a collision; there should be no collisions in + this file's code given how we expect to update elems. + + Warning: if this calls fails, the passed-in iterator can be rendered invalid + (and the elem erased from the container). + + Source: https://www.boost.org/latest/libs/multi_index/doc/tutorial/basics.html#ord_updating +*/ +template +static bool fly_tx_req(T &it, U &container, const uint64_t nonce) noexcept { + return container.modify(it, [nonce](tx_request &tx_req) { + tx_req.fly(nonce); + }); +} + +template +static bool start_processing_tx_req(T &it, U &container) noexcept { + return container.modify(it, [](tx_request &tx_req) { + tx_req.start_processing(); + }); +} + +uint64_t request_manager::bump_nonce(const boost::uuids::uuid &peer_id) { + std::lock_guard lock(m_mutex); + if (++m_connection_stats[peer_id].cur_nonce == 0) + ++m_connection_stats[peer_id].cur_nonce; + return m_connection_stats[peer_id].cur_nonce; +} + void request_manager::remove_peer(const boost::uuids::uuid &peer_id) { MINFO("Removing all requests for disconnected peer: " << epee::string_tools::pod_to_hex(peer_id)); std::lock_guard lock(m_mutex); @@ -59,7 +91,7 @@ std::unordered_set request_manager::remove_stale_requests() for (auto it = m_requested_txs.begin(); it != m_requested_txs.end();) { // Stale means it's been in flight for longer than the allowed timeout const auto elapsed = std::chrono::duration_cast(now - it->last_action_timestamp); - if (!it->in_flight || elapsed.count() < m_request_timeout) { + if (it->processing || !it->in_flight || elapsed.count() < m_request_timeout) { ++it; continue; } @@ -86,44 +118,55 @@ std::unordered_set request_manager::remove_stale_requests() return drop_peers_out; } -bool request_manager::add_request(const crypto::hash &tx_hash, const boost::uuids::uuid &peer_id) { - MINFO("Requesting from peer: " << epee::string_tools::pod_to_hex(peer_id) << " the transaction: " << tx_hash); +request_manager::tx_request_t request_manager::enqueue_requests(const std::vector &tx_hashes, const boost::uuids::uuid &peer_id) { std::lock_guard lock(m_mutex); - auto& by_peer_and_tx = get_requests_by_peer_and_tx(m_requested_txs); - auto it = by_peer_and_tx.find(boost::make_tuple(peer_id, tx_hash)); + tx_request_t tx_req(*this, peer_id); + for (const auto &tx_hash : tx_hashes) + { + MINFO("Requesting from peer: " << epee::string_tools::pod_to_hex(peer_id) << " the transaction: " << tx_hash); - const bool in_queue = it != by_peer_and_tx.end(); - const bool let_it_fly = m_connection_stats[peer_id].in_flight_requests < m_max_in_flight - && !this->request_is_in_flight(tx_hash); + auto& by_peer_and_tx = get_requests_by_peer_and_tx(m_requested_txs); + auto it = by_peer_and_tx.find(boost::make_tuple(peer_id, tx_hash)); - if (in_queue && !let_it_fly) { - // already have this peer for this tx, we can't process additional reqs at this time - MDEBUG("Peer " << epee::string_tools::pod_to_hex(peer_id) - << " already in request queue for tx " << tx_hash); - return false; - } + const bool in_queue = it != by_peer_and_tx.end(); + const bool let_it_fly = m_connection_stats[peer_id].in_flight_requests < m_max_in_flight + && !this->request_is_in_flight(tx_hash); - if (!in_queue) { - // Add the new request! - m_requested_txs.insert(tx_request(peer_id, tx_hash, let_it_fly)); - } else if (let_it_fly) { - // let_it_fly should always be true here - it->fly(); - } + if (!let_it_fly) { + if (!in_queue) { + // Add the new request! Nonce doesn't get set until it's in flight. + m_requested_txs.insert(tx_request(peer_id, tx_hash, 0/*nonce*/, let_it_fly)); + } else { + // already have this peer for this tx, but we can't process additional reqs at this time + MDEBUG("Peer " << epee::string_tools::pod_to_hex(peer_id) + << " already in request queue for tx " << tx_hash); + } + continue; + } + + // let_it_fly is true + assert(let_it_fly); + if (!in_queue) { + m_requested_txs.insert(tx_request(peer_id, tx_hash, tx_req.nonce, let_it_fly)); + } else { + const bool r = fly_tx_req(it, by_peer_and_tx, tx_req.nonce); + CHECK_AND_ASSERT_MES(r, tx_req, "Failed to fly tx request for tx " << tx_hash); + } - if (let_it_fly) { ++m_connection_stats[peer_id].n_total_requests; ++m_connection_stats[peer_id].in_flight_requests; + + tx_req.tx_hashes.push_back(tx_hash); } - return let_it_fly; + return tx_req; } -std::vector request_manager::fly_available_requests(const boost::uuids::uuid &peer_id) { +request_manager::tx_request_t request_manager::fly_available_requests(const boost::uuids::uuid &peer_id) { std::lock_guard lock(m_mutex); - std::vector tx_hashes; + tx_request_t tx_req(*this, peer_id); auto& by_peer = get_requests_by_peer_id(m_requested_txs); auto range = by_peer.equal_range(peer_id); for (auto it = range.first; it != range.second; ++it) @@ -133,14 +176,27 @@ std::vector request_manager::fly_available_requests(const boost::u if (it->in_flight || this->request_is_in_flight(it->tx_hash)) continue; - it->fly(); + const bool r = fly_tx_req(it, by_peer, tx_req.nonce); + CHECK_AND_ASSERT_MES(r, tx_req, "Failed to fly available tx request"); ++m_connection_stats[peer_id].n_total_requests; ++m_connection_stats[peer_id].in_flight_requests; - tx_hashes.push_back(it->tx_hash); + tx_req.tx_hashes.push_back(it->tx_hash); } - return tx_hashes; + return tx_req; +} + +template +T request_manager::erase_tx_request(T &it, U &container) { + CHECK_AND_ASSERT_MES(it != container.end(), container.end(), "Expected found tx request"); + const boost::uuids::uuid &peer_id = it->peer_id; + if (it->in_flight && m_connection_stats[peer_id].in_flight_requests > 0) + { + --m_connection_stats[peer_id].in_flight_requests; + MINFO("Decremented in_flight_requests count for peer: " << epee::string_tools::pod_to_hex(peer_id) << ", current in_flight_requests: " << m_connection_stats[peer_id].in_flight_requests); + } + return container.erase(it); } bool request_manager::remove_request(const crypto::hash &tx_hash) { @@ -152,18 +208,47 @@ bool request_manager::remove_request(const crypto::hash &tx_hash) { return false; } for (auto it = range.first; it != range.second;) { - const boost::uuids::uuid &peer_id = it->peer_id; - MDEBUG("Removing tx request " << it->tx_hash << " for peer " << epee::string_tools::pod_to_hex(peer_id)); - if (it->in_flight && m_connection_stats[peer_id].in_flight_requests > 0) - { - --m_connection_stats[peer_id].in_flight_requests; - MINFO("Decremented in_flight_requests count for peer: " << epee::string_tools::pod_to_hex(peer_id) << ", current in_flight_requests: " << m_connection_stats[peer_id].in_flight_requests); - } - it = by_tx_hash.erase(it); + MDEBUG("Removing tx request " << it->tx_hash << " for peer " << epee::string_tools::pod_to_hex(it->peer_id)); + it = erase_tx_request(it, by_tx_hash); } return true; } +void request_manager::processing_txs(const boost::uuids::uuid &peer_id, const uint64_t nonce, const std::vector &tx_hashes) { + // Any txs we requested are expected to have nonce >0 + if (nonce == 0) + return; + std::lock_guard lock(m_mutex); + // If no reqs found for that peer id and nonce, then we didn't request these hashes + auto& by_peer_and_nonce = get_requests_by_peer_and_nonce(m_requested_txs); + auto range = by_peer_and_nonce.equal_range(boost::make_tuple(peer_id, nonce)); + if (range.first == range.second) + return; + + // Peer shouldn't have included *more* txs in the response than can be requested at once. + CHECK_AND_ASSERT_MES(tx_hashes.size() <= m_max_in_flight,, "Too many tx_hashes in response"); + const std::unordered_set peer_hashes(tx_hashes.begin(), tx_hashes.end()); + + MINFO("Processing " << tx_hashes.size() << " txs received from peer " << epee::string_tools::pod_to_hex(peer_id) + << " (nonce=" << nonce << ")"); + + for (auto it = range.first; it != range.second;) { + const auto peer_hash_it = peer_hashes.find(it->tx_hash); + if (peer_hash_it == peer_hashes.end()) { + // If we requested a tx at this nonce, but it wasn't included in the peer's resp, then the peer is saying they + // didn't have the tx in their pool. We can remove it from our request queue for that specific peer. + MDEBUG("Removing tx request " << it->tx_hash << " only for peer " << epee::string_tools::pod_to_hex(it->peer_id)); + it = erase_tx_request(it, by_peer_and_nonce); + continue; + } + + // We requested the tx from the peer, the peer responded with it, and now we're going to process it. + const bool r = start_processing_tx_req(it, by_peer_and_nonce); + CHECK_AND_ASSERT_MES(r,, "Failed to mark tx " << *peer_hash_it << " as processing"); + ++it; + } +} + bool request_manager::missed_request(const boost::uuids::uuid &peer_id, const std::size_t n_missed_reqs) { std::lock_guard lock(m_mutex); if ((m_connection_stats[peer_id].missed + n_missed_reqs) > m_connection_stats[peer_id].missed) diff --git a/src/cryptonote_protocol/request_manager.h b/src/cryptonote_protocol/request_manager.h index 5c92b0d34aa..0a5dc866d31 100644 --- a/src/cryptonote_protocol/request_manager.h +++ b/src/cryptonote_protocol/request_manager.h @@ -65,6 +65,7 @@ class request_manager { private: struct connection_statistics { + std::atomic cur_nonce = 0; std::atomic n_total_requests = 0; std::atomic missed = 0; std::atomic in_flight_requests = 0; @@ -72,6 +73,16 @@ class request_manager { std::unordered_map m_connection_stats; public: + struct tx_request_t + { + const uint64_t nonce; + std::vector tx_hashes; + + tx_request_t(request_manager& rm, const boost::uuids::uuid &peer_id) + : nonce(rm.bump_nonce(peer_id)) + {}; + }; + request_manager(const std::size_t max_in_flight, const int64_t request_timeout = P2P_DEFAULT_REQUEST_TIMEOUT) : m_requested_txs(), @@ -86,17 +97,32 @@ class request_manager { // Returns the set of peers to drop because they've missed too many requests std::unordered_set remove_stale_requests(); - // Return true if the request should be sent over the connection - bool add_request(const crypto::hash &tx_hash, const boost::uuids::uuid &id); + // Return the tx_request_t object that should be immediately sent over the wire + // The function internally adds all passed tx_hashes to the request manager queue for the + // given peer, but may only need to request some of them. + tx_request_t enqueue_requests(const std::vector &tx_hashes, const boost::uuids::uuid &id); // Remove current in-flight request for a transaction, if present // true: found, false: not found or none in-flight bool remove_request(const crypto::hash &tx_hash); - // Returns the vector of tx hashes to request - std::vector fly_available_requests(const boost::uuids::uuid &peer_id); + // Processing received txs from peer. This function does 2 things: + // 1) Marks requested txs as processing. + // 2) If some tx hashes are missing that were requested at the given peer and nonce, removes those reqs from the queue. + void processing_txs(const boost::uuids::uuid &peer_id, const uint64_t nonce, const std::vector &tx_hashes); + + // Return the tx_request_t object that should be immediately sent over the wire + tx_request_t fly_available_requests(const boost::uuids::uuid &peer_id); private: + // Bump the tx request nonce for that peer + uint64_t bump_nonce(const boost::uuids::uuid &peer_id); + + // Removes a tx request that has already been found in the container + // Returns the iterator pointing to the next elem in the container + template + T erase_tx_request(T &it, U &container); + // Return true if we should drop the peer because it exceeded threshold for allowed missed reqs bool missed_request(const boost::uuids::uuid &peer_id, const std::size_t n_missed_reqs = 1); diff --git a/src/cryptonote_protocol/txrequestqueue.h b/src/cryptonote_protocol/txrequestqueue.h index bd70352ad14..7294d203ea6 100644 --- a/src/cryptonote_protocol/txrequestqueue.h +++ b/src/cryptonote_protocol/txrequestqueue.h @@ -53,22 +53,36 @@ struct tx_request { - boost::uuids::uuid peer_id; - crypto::hash tx_hash; - mutable std::chrono::steady_clock::time_point last_action_timestamp; - mutable bool in_flight = false; + /*const*/ boost::uuids::uuid peer_id; + /*const*/ crypto::hash tx_hash; + + uint64_t nonce = 0; + bool in_flight = false; + bool processing = false; + std::chrono::steady_clock::time_point last_action_timestamp; tx_request(const boost::uuids::uuid& _peer_id, const crypto::hash& _tx_hash, + const uint64_t _nonce, const bool _in_flight): peer_id(_peer_id), tx_hash(_tx_hash), + nonce(_nonce), last_action_timestamp(std::chrono::steady_clock::now()), in_flight(_in_flight) {} public: - void fly() const { in_flight = true; last_action_timestamp = std::chrono::steady_clock::now(); }; + void fly(const uint64_t _nonce) noexcept { + nonce = _nonce; + in_flight = true; + last_action_timestamp = std::chrono::steady_clock::now(); + } + + void start_processing() noexcept { + processing = true; + last_action_timestamp = std::chrono::steady_clock::now(); + } }; using boost::multi_index::hashed_non_unique; @@ -85,10 +99,15 @@ typedef multi_index_container< hashed_non_unique>, // Index 1: by tx_hash - all requests for a tx hashed_non_unique>, - // Index 2: by (peer_id, tx_hash) - unique requests + // Index 2: by (peer_id, tx_hash) - unique txs requested hashed_unique, member + >>, + // Index 3: by (peer_id, nonce) - all requests for a given peer and nonce + hashed_non_unique, + member >> > > request_container; @@ -108,4 +127,9 @@ decltype(auto) get_requests_by_peer_and_tx(container_t&& container) { return std::forward(container).template get<2>(); } +template +decltype(auto) get_requests_by_peer_and_nonce(container_t&& container) { + return std::forward(container).template get<3>(); +} + #endif // CRYPTONOTE_PROTOCOL_TXREQUESTQUEUE_H diff --git a/src/rpc/core_rpc_server.cpp b/src/rpc/core_rpc_server.cpp index 309cc767425..c2b49e577a6 100644 --- a/src/rpc/core_rpc_server.cpp +++ b/src/rpc/core_rpc_server.cpp @@ -1443,11 +1443,19 @@ namespace cryptonote } res.sanity_check_failed = false; + transaction tx{}; crypto::hash txid{}; + if (!parse_and_validate_tx_from_blob(tx_blob, tx, txid, true)) + { + res.status = "Failed"; + res.reason = "Failed to parse tx"; + return true; + } + if (!skip_validation) { tx_verification_context tvc{}; - if(!m_core.handle_incoming_tx(tx_blob, tvc, (req.do_not_relay ? relay_method::none : relay_method::local), false, txid) || tvc.m_verifivation_failed) + if(!m_core.handle_incoming_tx(tx_blob, tx, txid, tvc, (req.do_not_relay ? relay_method::none : relay_method::local), false) || tvc.m_verifivation_failed) { res.status = "Failed"; std::string reason = ""; @@ -1492,16 +1500,6 @@ namespace cryptonote return true; } } - else - { - transaction tx; - if (!parse_and_validate_tx_from_blob(tx_blob, tx, txid)) - { - res.status = "Failed"; - res.reason = "Failed to parse tx"; - return true; - } - } NOTIFY_NEW_TRANSACTIONS::request r; r.txs.push_back(std::move(tx_blob)); diff --git a/src/rpc/daemon_handler.cpp b/src/rpc/daemon_handler.cpp index fa0e2b85a83..6a3ecd31e04 100644 --- a/src/rpc/daemon_handler.cpp +++ b/src/rpc/daemon_handler.cpp @@ -397,8 +397,11 @@ namespace rpc tx_verification_context tvc = AUTO_VAL_INIT(tvc); + cryptonote::transaction tx{}; crypto::hash txid; - if(!m_core.handle_incoming_tx(tx_blob, tvc, (relay ? relay_method::local : relay_method::none), false, txid) || tvc.m_verifivation_failed) + if(!parse_and_validate_tx_from_blob(tx_blob, tx, txid, true) + || !m_core.handle_incoming_tx(tx_blob, tx, txid, tvc, (relay ? relay_method::local : relay_method::none), false) + || tvc.m_verifivation_failed) { if (tvc.m_verifivation_failed) { diff --git a/tests/core_tests/chaingen.h b/tests/core_tests/chaingen.h index 66d5826914f..ab0664b2013 100644 --- a/tests/core_tests/chaingen.h +++ b/tests/core_tests/chaingen.h @@ -599,8 +599,13 @@ struct push_core_event_visitor: public boost::static_visitor cryptonote::tx_verification_context tvc = AUTO_VAL_INIT(tvc); size_t pool_size = m_c.get_pool_transactions_count(); - crypto::hash txid; - m_c.handle_incoming_tx(t_serializable_object_to_blob(tx), tvc, m_tx_relay, false, txid); + const cryptonote::blobdata tx_blob = t_serializable_object_to_blob(tx); + cryptonote::transaction reparsed_tx{}; + crypto::hash txid{}; + if (cryptonote::parse_and_validate_tx_from_blob(tx_blob, reparsed_tx, txid)) + m_c.handle_incoming_tx(tx_blob, reparsed_tx, txid, tvc, m_tx_relay, false); + else + tvc.m_verifivation_failed = true; bool tx_added = pool_size + 1 == m_c.get_pool_transactions_count(); bool r = m_validator.check_tx_verification_context(tvc, tx_added, m_ev_index, tx); CHECK_AND_NO_ASSERT_MES(r, false, "tx verification context check failed"); @@ -622,8 +627,12 @@ struct push_core_event_visitor: public boost::static_visitor size_t pool_size = m_c.get_pool_transactions_count(); for (size_t i = 0; i < tx_blobs.size(); ++i) { - crypto::hash txid; - m_c.handle_incoming_tx(tx_blobs[i], tvcs[i], m_tx_relay, false, txid); + cryptonote::transaction tx{}; + crypto::hash txid{}; + if (cryptonote::parse_and_validate_tx_from_blob(tx_blobs[i], tx, txid)) + m_c.handle_incoming_tx(tx_blobs[i], tx, txid, tvcs[i], m_tx_relay, false); + else + tvcs[i].m_verifivation_failed = true; } size_t tx_added = m_c.get_pool_transactions_count() - pool_size; bool r = m_validator.check_tx_verification_context_array(tvcs, tx_added, m_ev_index, txs); @@ -702,18 +711,13 @@ struct push_core_event_visitor: public boost::static_visitor cryptonote::tx_verification_context tvc = AUTO_VAL_INIT(tvc); size_t pool_size = m_c.get_pool_transactions_count(); + cryptonote::transaction tx{}; crypto::hash txid; - m_c.handle_incoming_tx(sr_tx.data, tvc, m_tx_relay, false, txid); + if (cryptonote::parse_and_validate_tx_from_blob(sr_tx.data, tx, txid)) + m_c.handle_incoming_tx(sr_tx.data, tx, txid, tvc, m_tx_relay, false); + else + tvc.m_verifivation_failed = true; bool tx_added = pool_size + 1 == m_c.get_pool_transactions_count(); - - cryptonote::transaction tx; - binary_archive ba{epee::strspan(sr_tx.data)}; - ::serialization::serialize(ba, tx); - if (!ba.good()) - { - tx = cryptonote::transaction(); - } - bool r = m_validator.check_tx_verification_context(tvc, tx_added, m_ev_index, tx); CHECK_AND_NO_ASSERT_MES(r, false, "transaction verification context check failed"); return true; diff --git a/tests/fuzz/fuzz_rpc/initialisation.cpp b/tests/fuzz/fuzz_rpc/initialisation.cpp index 7d2396471a2..153fce0d8c7 100644 --- a/tests/fuzz/fuzz_rpc/initialisation.cpp +++ b/tests/fuzz/fuzz_rpc/initialisation.cpp @@ -300,16 +300,18 @@ bool generate_random_blocks(cryptonote::core& core, FuzzedDataProvider& provider } for (const auto& tx_blob : cached_txs) { - cryptonote::tx_verification_context tvc; + cryptonote::transaction tx{}; crypto::hash tx_hash; - bool accepted = core.handle_incoming_tx(tx_blob, tvc, cryptonote::relay_method::block, true, tx_hash); - if (accepted || tvc.m_added_to_pool) { - // Store legit hashes - cryptonote::transaction tx; - if (cryptonote::parse_and_validate_tx_from_blob(tx_blob, tx)) { - cached_tx_hashes.push_back(cryptonote::get_transaction_hash(tx)); - } + if (!cryptonote::parse_and_validate_tx_from_blob(tx_blob, tx, tx_hash)) + continue; + cryptonote::tx_verification_context tvc{}; + if (!core.handle_incoming_tx(tx_blob, tx, tx_hash, tvc, cryptonote::relay_method::block, true) + && !tvc.m_added_to_pool) + { + continue; } + // Store legit hashes + cached_tx_hashes.push_back(tx_hash); } return added_block; diff --git a/tests/unit_tests/node_server.cpp b/tests/unit_tests/node_server.cpp index 826e34332ef..4ac0099c84b 100644 --- a/tests/unit_tests/node_server.cpp +++ b/tests/unit_tests/node_server.cpp @@ -80,7 +80,7 @@ class test_core : public cryptonote::i_core_events bool have_block(const crypto::hash& id, int *where = NULL) const {return false;} bool have_block_unlocked(const crypto::hash& id, int *where = NULL) const {return false;} void get_blockchain_top(uint64_t& height, crypto::hash& top_id)const{height=0;top_id=crypto::null_hash;} - bool handle_incoming_tx(const cryptonote::blobdata& tx_blob, cryptonote::tx_verification_context& tvc, cryptonote::relay_method tx_relay, bool relayed, crypto::hash& txid) { return true; } + bool handle_incoming_tx(const cryptonote::blobdata& tx_blob, cryptonote::transaction& tx, crypto::hash& txid, cryptonote::tx_verification_context& tvc, cryptonote::relay_method tx_relay, bool relayed) { return true; } bool handle_single_incoming_block(const cryptonote::blobdata& block_blob, const cryptonote::block *b, cryptonote::block_verification_context& bvc, cryptonote::pool_supplement& extra_block_txs, bool update_miner_blocktemplate = true) { return true; } bool handle_incoming_block(const cryptonote::blobdata& block_blob, const cryptonote::block *block, cryptonote::block_verification_context& bvc, bool update_miner_blocktemplate = true) { return true; } bool handle_incoming_block(const cryptonote::blobdata& block_blob, const cryptonote::block *block, cryptonote::block_verification_context& bvc, cryptonote::pool_supplement& extra_block_txs, bool update_miner_blocktemplate = true) { return true; } diff --git a/tests/unit_tests/request_manager.cpp b/tests/unit_tests/request_manager.cpp index 4eabac6c49b..28f112ffbcc 100644 --- a/tests/unit_tests/request_manager.cpp +++ b/tests/unit_tests/request_manager.cpp @@ -53,21 +53,21 @@ TEST(request_manager, basic_usage) for (char i = 0; i < MAX_IN_FLIGHT; ++i) { // Success case - ASSERT_TRUE(req_manager.add_request(crypto::hash{i}, peer)); + ASSERT_EQ(req_manager.enqueue_requests({crypto::hash{i}}, peer).tx_hashes.size(), 1); // Can't re-add - ASSERT_FALSE(req_manager.add_request(crypto::hash{i}, peer)); + ASSERT_EQ(req_manager.enqueue_requests({crypto::hash{i}}, peer).tx_hashes.size(), 0); } // 2. Queue is full and can't let any others fly. Queue this one - ASSERT_FALSE(req_manager.add_request(crypto::hash{MAX_IN_FLIGHT}, peer)); + ASSERT_EQ(req_manager.enqueue_requests({crypto::hash{MAX_IN_FLIGHT}}, peer).tx_hashes.size(), 0); // 3. Remove 1 from the queue to make room for one queued above ASSERT_TRUE(req_manager.remove_request(crypto::hash{0})); ASSERT_FALSE(req_manager.remove_request(crypto::hash{0})); // 4. Make sure the one queued above enters flight - const auto tx_reqs = req_manager.fly_available_requests(peer); + const auto tx_reqs = req_manager.fly_available_requests(peer).tx_hashes; ASSERT_EQ(tx_reqs.size(), 1); ASSERT_EQ(tx_reqs.front(), crypto::hash{MAX_IN_FLIGHT}); @@ -91,10 +91,10 @@ TEST(request_manager, multiple_peers) const boost::uuids::uuid peer = uuid_from_char(i); // Success case - ASSERT_TRUE(req_manager.add_request(hash, peer)); + ASSERT_EQ(req_manager.enqueue_requests({hash}, peer).tx_hashes.size(), 1); // Can't re-add - ASSERT_FALSE(req_manager.add_request(hash, peer)); + ASSERT_EQ(req_manager.enqueue_requests({hash}, peer).tx_hashes.size(), 0); } // 2. No stale requests and none available @@ -104,7 +104,7 @@ TEST(request_manager, multiple_peers) for (uint8_t i = 0; i < MAX_IN_FLIGHT; ++i) { const boost::uuids::uuid peer = uuid_from_char(i); - const auto tx_hashes = req_manager.fly_available_requests(peer); + const auto tx_hashes = req_manager.fly_available_requests(peer).tx_hashes; ASSERT_TRUE(tx_hashes.empty()); } } @@ -119,19 +119,19 @@ TEST(request_manager, multiple_peers) { const crypto::hash hash{i}; const boost::uuids::uuid peer = uuid_from_char(MAX_IN_FLIGHT - i - 1); - ASSERT_FALSE(req_manager.add_request(hash, peer)); + ASSERT_EQ(req_manager.enqueue_requests({hash}, peer).tx_hashes.size(), 0); } // 5. Remove stale requests const auto drop_peers = req_manager.remove_stale_requests(); ASSERT_TRUE(drop_peers.empty()); - // 6. Available reqs should be all the tx hashes from step 3 + // 6. Available reqs should be all the tx hashes from step 4 for (char i = 0; i < MAX_IN_FLIGHT; ++i) { const crypto::hash hash{i}; const boost::uuids::uuid peer = uuid_from_char(MAX_IN_FLIGHT - i - 1); - const auto tx_hashes = req_manager.fly_available_requests(peer); + const auto tx_hashes = req_manager.fly_available_requests(peer).tx_hashes; ASSERT_EQ(tx_hashes.size(), 1); ASSERT_EQ(tx_hashes.front(), hash); @@ -155,7 +155,7 @@ TEST(request_manager, drop_peers) // 1. Add many requests for (char i = 0; i < MAX_IN_FLIGHT; ++i) - ASSERT_TRUE(req_manager.add_request(crypto::hash{i}, peer)); + ASSERT_EQ(req_manager.enqueue_requests({crypto::hash{i}}, peer).tx_hashes.size(), 1); // 2. Sleep to let requests timeout MINFO("Sleeping for 20ms to let requests timeout"); @@ -170,4 +170,86 @@ TEST(request_manager, drop_peers) req_manager.remove_peer(peer); } //---------------------------------------------------------------------------------------------------------------------- +TEST(request_manager, dont_rm_processing_txs) +{ + const uint8_t MAX_IN_FLIGHT = P2P_MIN_SAMPLE_SIZE_FOR_DROPPING; + const int64_t TIMEOUT_MS = 10; + request_manager req_manager(MAX_IN_FLIGHT, TIMEOUT_MS); + + const boost::uuids::uuid peer{}; + + // 1. Add many requests + std::vector init_hashes; + init_hashes.reserve(MAX_IN_FLIGHT); + for (char i = 0; i < MAX_IN_FLIGHT; ++i) + init_hashes.emplace_back(crypto::hash{i}); + const auto tx_req = req_manager.enqueue_requests(init_hashes, peer); + ASSERT_EQ(tx_req.tx_hashes, init_hashes); + ASSERT_EQ(tx_req.nonce, 1); + + // 2. Indicate that we're processing all of the tx requests + // Note: the last one is explicitly not included in the processing_hashes, which technically is supposed to mean + // the peer didn't have the tx and it can be removed from the request queue. + const std::vector processing_hashes(init_hashes.begin(), init_hashes.end() - 1); + req_manager.processing_txs(peer, tx_req.nonce, processing_hashes); + ASSERT_FALSE(req_manager.remove_request(init_hashes.back())); + + // 3. Sleep to let requests timeout + MINFO("Sleeping for 20ms to let requests timeout"); + usleep(TIMEOUT_MS * 1000 * 2); + + // 4. There should be no stale requests. + const auto drop_peers = req_manager.remove_stale_requests(); + ASSERT_EQ(drop_peers.size(), 0); + + // 5. Remove peers + req_manager.remove_peer(peer); +} +//---------------------------------------------------------------------------------------------------------------------- +TEST(request_manager, dont_rm_processing_txs_enqueue_overage) +{ + const uint8_t MAX_IN_FLIGHT = 2; + const int64_t TIMEOUT_MS = 10; + request_manager req_manager(MAX_IN_FLIGHT, TIMEOUT_MS); + + const boost::uuids::uuid peer{}; + + // 1. Enqueue 1 more request than max allowed in flight + const uint8_t TOTAL_TXS = MAX_IN_FLIGHT + 1; + std::vector init_hashes; + init_hashes.reserve(TOTAL_TXS); + for (char i = 0; i < TOTAL_TXS; ++i) + init_hashes.emplace_back(crypto::hash{i}); + const auto tx_req = req_manager.enqueue_requests(init_hashes, peer); + ASSERT_EQ(tx_req.tx_hashes, std::vector(init_hashes.begin(), init_hashes.begin() + MAX_IN_FLIGHT)); + ASSERT_EQ(tx_req.nonce, 1); + + // 2. Remove 1 req to make room for the last one + ASSERT_TRUE(req_manager.remove_request(init_hashes.front())); + + // 3. Now fly the last one + const auto tx_req2 = req_manager.fly_available_requests(peer); + ASSERT_EQ(tx_req2.tx_hashes, std::vector{init_hashes.back()}); + ASSERT_EQ(tx_req2.nonce, 2); + + // 4. Indicate that we're processing Step 3's tx + req_manager.processing_txs(peer, tx_req2.nonce, tx_req2.tx_hashes); + + // 5. Sleep to let requests timeout + MINFO("Sleeping for 20ms to let requests timeout"); + usleep(TIMEOUT_MS * 1000 * 2); + + // 6. The processing tx from step 4 should still be present, since it's processing and shouldn't have been rm'd. + // init_hashes.at(1) should have timed out and rm'd. Since this is only the first stale tx req rm'd, the peer + // is not expected to drop. + static_assert(P2P_MIN_SAMPLE_SIZE_FOR_DROPPING > 1); + const auto drop_peers = req_manager.remove_stale_requests(); + ASSERT_EQ(drop_peers.size(), 0); + ASSERT_FALSE(req_manager.remove_request(init_hashes.at(1))); + ASSERT_TRUE(req_manager.remove_request(init_hashes.back())); + + // 7. Remove peer + req_manager.remove_peer(peer); +} +//---------------------------------------------------------------------------------------------------------------------- //----------------------------------------------------------------------------------------------------------------------