From 45821d078a998925c58e8ce7fc2a7c1c5a58bbd3 Mon Sep 17 00:00:00 2001 From: PastaClaw Date: Tue, 14 Jul 2026 22:43:48 -0500 Subject: [PATCH] perf: deserialize DKG network messages once Keep incoming DKG messages as exact raw wire bytes after cheap framing validation. The DKG worker remains the sole normal typed and BLS deserialization point, avoiding repeated point decoding on the message-handler thread. Preserve authenticated-only intake, parameter-derived size and framing checks, exact-wire inventory hashes, duplicate-before-quota ordering, worker preverification, malformed-message scoring in the matching phase, and own-message processing. Bound each raw message-type queue across all remote NodeIds to four times the quorum size, allowing two peers to use their full per-node quotas, with one reserved slot for the local phase message. New-round cleanup discards bounded stale raw bytes without deserializing them, so reconnect churn cannot create an unbounded synchronous BLS workload before quorum initialization. Add unit coverage for per-node and queue-wide caps, duplicate ordering, local-message capacity, and round clearing. Update the focused functional test to cover worker-deferred BLS rejection plus bounded late-message retention across reconnect-generated NodeIds and prompt next-round initialization. Co-Authored-By: Claude --- src/llmq/dkgsessionhandler.cpp | 38 ++- src/llmq/dkgsessionhandler.h | 44 +-- src/llmq/net_dkg.cpp | 308 +++++++++++++++++---- src/test/llmq_dkg_tests.cpp | 68 +++++ test/functional/feature_llmq_dkg_intake.py | 150 +++++++++- 5 files changed, 494 insertions(+), 114 deletions(-) diff --git a/src/llmq/dkgsessionhandler.cpp b/src/llmq/dkgsessionhandler.cpp index c9258ff353f9..b656f532e80e 100644 --- a/src/llmq/dkgsessionhandler.cpp +++ b/src/llmq/dkgsessionhandler.cpp @@ -29,18 +29,32 @@ void CDKGPendingMessages::PushPendingMessage(NodeId from, std::shared_ptr= maxMessagesPerNode) { + // Check duplicates before charging the per-node quota so a peer that + // resends the same hash cannot exhaust its budget with dupes. + if (seenMessages.count(hash) != 0) { + LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from); + return; + } + + const auto node_it = messagesPerNode.find(from); + if (node_it != messagesPerNode.end() && node_it->second >= maxMessagesPerNode) { // TODO ban? LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- too many messages, peer=%d\n", __func__, from); return; } - messagesPerNode[from]++; - if (!seenMessages.emplace(hash).second) { - LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from); + const bool is_remote = from != -1; + if ((is_remote && pendingRemoteMessageCount >= maxPendingRemoteMessages) || + pendingMessages.size() >= maxPendingMessages) { + LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- pending queue full, peer=%d\n", __func__, from); return; } + messagesPerNode[from]++; + if (is_remote) { + pendingRemoteMessageCount++; + } + seenMessages.emplace(hash); pendingMessages.emplace_back(std::make_pair(from, std::move(pm))); } @@ -50,6 +64,9 @@ std::list CDKGPendingMessages::PopPendingMes std::list ret; while (!pendingMessages.empty() && ret.size() < maxCount) { + if (pendingMessages.front().first != -1) { + pendingRemoteMessageCount--; + } ret.emplace_back(std::move(pendingMessages.front())); pendingMessages.pop_front(); } @@ -57,20 +74,21 @@ std::list CDKGPendingMessages::PopPendingMes return ret; } -bool CDKGPendingMessages::HasSeen(const uint256& hash) const -{ - LOCK(cs_messages); - return seenMessages.count(hash) != 0; -} - void CDKGPendingMessages::Clear() { LOCK(cs_messages); pendingMessages.clear(); + pendingRemoteMessageCount = 0; messagesPerNode.clear(); seenMessages.clear(); } +bool CDKGPendingMessages::HasSeen(const uint256& hash) const +{ + LOCK(cs_messages); + return seenMessages.count(hash) != 0; +} + void CDKGSessionHandler::ClearPendingMessages() { pendingContributions.Clear(); diff --git a/src/llmq/dkgsessionhandler.h b/src/llmq/dkgsessionhandler.h index be55bfcbaa8a..c631f103126d 100644 --- a/src/llmq/dkgsessionhandler.h +++ b/src/llmq/dkgsessionhandler.h @@ -11,8 +11,6 @@ #include #include #include -#include -#include #include class CDataStream; @@ -47,7 +45,7 @@ enum class QuorumPhase { * main handler thread, we push them into a CDKGPendingMessages object and later pop+deserialize them in the DKG phase * handler thread. * - * Each message type has it's own instance of this class. + * Each message type has its own instance of this class. */ class CDKGPendingMessages { @@ -56,20 +54,31 @@ class CDKGPendingMessages private: const size_t maxMessagesPerNode; + const size_t maxPendingRemoteMessages; + const size_t maxPendingMessages; mutable Mutex cs_messages; std::list pendingMessages GUARDED_BY(cs_messages); + size_t pendingRemoteMessageCount GUARDED_BY(cs_messages){0}; std::map messagesPerNode GUARDED_BY(cs_messages); Uint256HashSet seenMessages GUARDED_BY(cs_messages); public: explicit CDKGPendingMessages(size_t _maxMessagesPerNode) : - maxMessagesPerNode(_maxMessagesPerNode) {}; + maxMessagesPerNode(_maxMessagesPerNode), + // Let two peers use their full quota while keeping reconnect-generated + // NodeIds from growing the queue without bound. + maxPendingRemoteMessages(_maxMessagesPerNode * 2), + // Reserve one slot for the message produced by this node during the + // matching phase. + maxPendingMessages(maxPendingRemoteMessages + 1) + { + } /** * Enqueue a serialized DKG message under @p from with content hash @p hash. * Caller is responsible for hashing the payload and (for real peers) * routing the erase-request to PeerManager. Drops the message silently on - * per-node capacity overflow or duplicate hash. + * per-node or queue-wide capacity overflow, or duplicate hash. */ void PushPendingMessage(NodeId from, std::shared_ptr pm, const uint256& hash) EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); @@ -77,31 +86,6 @@ class CDKGPendingMessages std::list PopPendingMessages(size_t maxCount) EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); bool HasSeen(const uint256& hash) const EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); void Clear() EXCLUSIVE_LOCKS_REQUIRED(!cs_messages); - - // Might return nullptr messages, which indicates that deserialization failed for some reason - template - std::vector>> PopAndDeserializeMessages(size_t maxCount) - EXCLUSIVE_LOCKS_REQUIRED(!cs_messages) - { - auto binaryMessages = PopPendingMessages(maxCount); - if (binaryMessages.empty()) { - return {}; - } - - std::vector>> ret; - ret.reserve(binaryMessages.size()); - for (const auto& bm : binaryMessages) { - auto msg = std::make_shared(); - try { - *bm.second >> *msg; - } catch (...) { - msg = nullptr; - } - ret.emplace_back(std::make_pair(bm.first, std::move(msg))); - } - - return ret; - } }; /** diff --git a/src/llmq/net_dkg.cpp b/src/llmq/net_dkg.cpp index 2d9bf31d01ab..6a899ea179e8 100644 --- a/src/llmq/net_dkg.cpp +++ b/src/llmq/net_dkg.cpp @@ -71,46 +71,195 @@ size_t MaxDKGMessageSize(std::string_view msg_type, const Consensus::LLMQParams& return cap < HARD_CEILING ? cap : HARD_CEILING; } -// Cheap, param-only structural validation of a pushed DKG message, run at intake -// before retention. Deserializes a COPY of the payload (leaving the caller's bytes -// intact for the pending queue and its inventory hash) and checks only safe upper -// bounds derived from quorum params: no member-list lookup and no signature -// verification, which remain on the DKG worker thread. Deserializing the copy does -// decompress the BLS points carried in the payload, but that work is bounded by -// the size cap applied just before this check. Rejects malformed or clearly -// oversized payloads before retention. -bool CheckDKGMessageStructure(std::string_view msg_type, const CDataStream& vRecv, const Consensus::LLMQParams& params) +bool SkipBytes(CDataStream& ds, size_t size) +{ + try { + ds.ignore(size); + return true; + } catch (const std::exception&) { + return false; + } +} + +std::optional ReadCompactSizeNoThrow(CDataStream& ds) +{ + try { + return ReadCompactSize(ds); + } catch (const std::exception&) { + return std::nullopt; + } +} + +bool SkipCommonDKGFields(CDataStream& ds) +{ + constexpr size_t PREFIX = 1 + 32 + 32; // llmqType + quorumHash + proTxHash + return SkipBytes(ds, PREFIX); +} + +// BLS encodings have fixed wire sizes, so intake only needs to establish that +// the bytes are present. Decoding and canonical/scheme validation remain on the +// DKG worker, where each object is materialized exactly once. +template +bool SkipBLSObject(CDataStream& ds) +{ + return SkipBytes(ds, BLSObject::SerSize); +} + +bool ReadAndCheckDynBitset(CDataStream& ds, uint64_t max_size, uint64_t& size_ret) +{ + const auto size = ReadCompactSizeNoThrow(ds); + if (!size.has_value() || size.value() > max_size) { + return false; + } + + const size_t byte_size = (size.value() + 7) / 8; + std::vector bytes(byte_size); + try { + ds.read(AsWritableBytes(Span{bytes})); + } catch (const std::exception&) { + return false; + } + + if (!bytes.empty() && bytes.size() * 8 != size.value()) { + const size_t rem = bytes.size() * 8 - size.value(); + const uint8_t mask = ~(uint8_t)(0xff >> rem); + if (bytes.back() & mask) { + return false; + } + } + + size_ret = size.value(); + return true; +} + +bool CheckContributionWireStructure(CDataStream& ds, const Consensus::LLMQParams& params) { const size_t size = params.size > 0 ? static_cast(params.size) : 0; const size_t min_size = params.minSize > 0 ? static_cast(params.minSize) : 0; const size_t threshold = params.threshold > 0 ? static_cast(params.threshold) : 0; - try { - CDataStream s(vRecv); // copy; deserialization does not advance the caller's stream - if (msg_type == NetMsgType::QCONTRIB) { - CDKGContribution qc; - s >> qc; - return qc.vvec != nullptr && qc.vvec->size() == threshold && - qc.contributions != nullptr && - qc.contributions->blobs.size() >= min_size && - qc.contributions->blobs.size() <= size; - } else if (msg_type == NetMsgType::QCOMPLAINT) { - CDKGComplaint qc; - s >> qc; - return qc.badMembers.size() == qc.complainForMembers.size() && - qc.badMembers.size() <= size; - } else if (msg_type == NetMsgType::QJUSTIFICATION) { - CDKGJustification qj; - s >> qj; - return qj.contributions.size() <= size; - } else if (msg_type == NetMsgType::QPCOMMITMENT) { - CDKGPrematureCommitment qc; - s >> qc; - return qc.validMembers.size() <= size; + + if (!SkipCommonDKGFields(ds)) { + return false; + } + + const auto vvec_size = ReadCompactSizeNoThrow(ds); + if (!vvec_size.has_value() || vvec_size.value() != threshold) { + return false; + } + for (uint64_t i = 0; i < vvec_size.value(); ++i) { + if (!SkipBLSObject(ds)) { + return false; } + } + + if (!SkipBLSObject(ds) || !SkipBytes(ds, 32)) { return false; - } catch (const std::exception&) { + } + + const auto blob_count = ReadCompactSizeNoThrow(ds); + if (!blob_count.has_value() || blob_count.value() < min_size || blob_count.value() > size) { return false; } + for (uint64_t i = 0; i < blob_count.value(); ++i) { + const auto blob_size = ReadCompactSizeNoThrow(ds); + if (!blob_size.has_value() || !SkipBytes(ds, blob_size.value())) { + return false; + } + } + + return SkipBLSObject(ds) && ds.empty(); +} + +bool CheckComplaintWireStructure(CDataStream& ds, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + uint64_t bad_members_size{0}; + uint64_t complain_for_members_size{0}; + return SkipCommonDKGFields(ds) && ReadAndCheckDynBitset(ds, size, bad_members_size) && + ReadAndCheckDynBitset(ds, size, complain_for_members_size) && + bad_members_size == complain_for_members_size && SkipBLSObject(ds) && ds.empty(); +} + +bool CheckJustificationWireStructure(CDataStream& ds, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + if (!SkipCommonDKGFields(ds)) { + return false; + } + + const auto contribution_count = ReadCompactSizeNoThrow(ds); + if (!contribution_count.has_value() || contribution_count.value() > size) { + return false; + } + for (uint64_t i = 0; i < contribution_count.value(); ++i) { + if (!SkipBytes(ds, 4) || !SkipBLSObject(ds)) { + return false; + } + } + + return SkipBLSObject(ds) && ds.empty(); +} + +bool CheckPrematureCommitmentWireStructure(CDataStream& ds, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + uint64_t valid_members_size{0}; + return SkipCommonDKGFields(ds) && ReadAndCheckDynBitset(ds, size, valid_members_size) && + SkipBLSObject(ds) && SkipBytes(ds, 32) && SkipBLSObject(ds) && + SkipBLSObject(ds) && ds.empty(); +} + +bool CheckDKGMessageWireStructure(std::string_view msg_type, const CDataStream& payload, const Consensus::LLMQParams& params) +{ + CDataStream ds(payload); + if (msg_type == NetMsgType::QCONTRIB) { + return CheckContributionWireStructure(ds, params); + } else if (msg_type == NetMsgType::QCOMPLAINT) { + return CheckComplaintWireStructure(ds, params); + } else if (msg_type == NetMsgType::QJUSTIFICATION) { + return CheckJustificationWireStructure(ds, params); + } else if (msg_type == NetMsgType::QPCOMMITMENT) { + return CheckPrematureCommitmentWireStructure(ds, params); + } + return false; +} + +// Param-only structural validation of a typed DKG message: checks only safe +// upper bounds derived from quorum params. Called by the DKG worker immediately +// after deserializing queued bytes and before PreVerifyMessage, so malformed +// structures never reach deeper validation. +template +bool CheckDKGMessageStructure(const Message& msg, const Consensus::LLMQParams& params); + +template <> +bool CheckDKGMessageStructure(const CDKGContribution& qc, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + const size_t min_size = params.minSize > 0 ? static_cast(params.minSize) : 0; + const size_t threshold = params.threshold > 0 ? static_cast(params.threshold) : 0; + return qc.vvec != nullptr && qc.vvec->size() == threshold && qc.contributions != nullptr && + qc.contributions->blobs.size() >= min_size && qc.contributions->blobs.size() <= size; +} + +template <> +bool CheckDKGMessageStructure(const CDKGComplaint& qc, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + return qc.badMembers.size() == qc.complainForMembers.size() && qc.badMembers.size() <= size; +} + +template <> +bool CheckDKGMessageStructure(const CDKGJustification& qj, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + return qj.contributions.size() <= size; +} + +template <> +bool CheckDKGMessageStructure(const CDKGPrematureCommitment& qc, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + return qc.validMembers.size() <= size; } // returns a set of NodeIds which sent invalid messages @@ -257,6 +406,10 @@ void RelayInvToParticipants(const CDKGSession& session, const CConnman& connman, template void EnqueueOwn(CDKGPendingMessages& pending, const Message& msg) { + // Own messages skip the wire path but still populate the pending queue so + // the DKG worker sees them alongside peer messages. The inventory hash is + // computed over the serialized form so it matches what remote peers would + // compute for the same message. CDataStream ds(SER_NETWORK, PROTOCOL_VERSION); ds << msg; auto pm = std::make_shared(std::move(ds)); @@ -266,10 +419,31 @@ void EnqueueOwn(CDKGPendingMessages& pending, const Message& msg) } template -bool ProcessPendingMessageBatch(const CConnman& connman, CDKGSession& session, CDKGPendingMessages& pendingMessages, - PeerManagerInternal& peerman, size_t maxCount) +bool DeserializeAndCheckDKGMessage(CDataStream& ds, const Consensus::LLMQParams& params, std::shared_ptr& msg) { - auto msgs = pendingMessages.PopAndDeserializeMessages(maxCount); + msg = std::make_shared(); + try { + ds >> *msg; + } catch (...) { + msg.reset(); + return false; + } + if (!ds.empty()) { + msg.reset(); + return false; + } + if (!CheckDKGMessageStructure(*msg, params)) { + msg.reset(); + return false; + } + return true; +} + +template +bool ProcessPendingMessageBatch(const CConnman& connman, CDKGSession& session, const Consensus::LLMQParams& params, + CDKGPendingMessages& pendingMessages, PeerManagerInternal& peerman, size_t maxCount) +{ + auto msgs = pendingMessages.PopPendingMessages(maxCount); if (msgs.empty()) { return false; } @@ -279,13 +453,14 @@ bool ProcessPendingMessageBatch(const CConnman& connman, CDKGSession& session, C for (const auto& p : msgs) { const NodeId& nodeId = p.first; - if (!p.second) { + std::shared_ptr msg; + if (!DeserializeAndCheckDKGMessage(*p.second, params, msg)) { LogPrint(BCLog::LLMQ_DKG, "%s -- failed to deserialize message, peer=%d\n", __func__, nodeId); peerman.PeerMisbehaving(nodeId, 100); continue; } bool ban = false; - if (!session.PreVerifyMessage(*p.second, ban)) { + if (!session.PreVerifyMessage(*msg, ban)) { if (ban) { LogPrint(BCLog::LLMQ_DKG, "%s -- banning node due to failed preverification, peer=%d\n", __func__, nodeId); peerman.PeerMisbehaving(nodeId, 100); @@ -293,7 +468,7 @@ bool ProcessPendingMessageBatch(const CConnman& connman, CDKGSession& session, C LogPrint(BCLog::LLMQ_DKG, "%s -- skipping message due to failed preverification, peer=%d\n", __func__, nodeId); continue; } - preverifiedMessages.emplace_back(p); + preverifiedMessages.emplace_back(nodeId, std::move(msg)); } if (preverifiedMessages.empty()) { return true; @@ -320,6 +495,7 @@ bool ProcessPendingMessageBatch(const CConnman& connman, CDKGSession& session, C return true; } + } // namespace @@ -403,10 +579,17 @@ void NetDKG::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStre Consensus::LLMQType llmqType; uint256 quorumHash; - vRecv >> llmqType; - vRecv >> quorumHash; - vRecv.Rewind(sizeof(uint256)); - vRecv.Rewind(sizeof(uint8_t)); + try { + vRecv >> llmqType; + vRecv >> quorumHash; + } catch (const std::exception&) { + m_peer_manager->PeerMisbehaving(pfrom.GetId(), 100, "malformed DKG message"); + return; + } + if (!vRecv.Rewind(sizeof(uint256)) || !vRecv.Rewind(sizeof(uint8_t))) { + m_peer_manager->PeerMisbehaving(pfrom.GetId(), 100, "malformed DKG message"); + return; + } const auto& llmq_params_opt = Params().GetLLMQ(llmqType); if (!llmq_params_opt.has_value()) { @@ -457,30 +640,30 @@ void NetDKG::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStre m_peer_manager->PeerMisbehaving(pfrom.GetId(), 100, "oversized DKG message"); return; } - - // Cheap structural pre-validation before retention. Validates a copy so the - // original bytes (and their inventory hash) are preserved for the worker. - if (!CheckDKGMessageStructure(msg_type, vRecv, llmq_params)) { + if (!CheckDKGMessageWireStructure(msg_type, vRecv, llmq_params)) { m_peer_manager->PeerMisbehaving(pfrom.GetId(), 100, "malformed DKG message"); return; } + // Inventory hash is computed over the raw wire bytes before we consume + // them, matching what a peer would compute for the same payload. + CHashWriter hw(SER_GETHASH, 0); + hw.write(AsWritableBytes(Span{vRecv})); + const uint256 hash = hw.GetHash(); + int inv_type = 0; - if (msg_type == NetMsgType::QCONTRIB) + if (msg_type == NetMsgType::QCONTRIB) { inv_type = MSG_QUORUM_CONTRIB; - else if (msg_type == NetMsgType::QCOMPLAINT) + } else if (msg_type == NetMsgType::QCOMPLAINT) { inv_type = MSG_QUORUM_COMPLAINT; - else if (msg_type == NetMsgType::QJUSTIFICATION) + } else if (msg_type == NetMsgType::QJUSTIFICATION) { inv_type = MSG_QUORUM_JUSTIFICATION; - else if (msg_type == NetMsgType::QPCOMMITMENT) + } else if (msg_type == NetMsgType::QPCOMMITMENT) { inv_type = MSG_QUORUM_PREMATURE_COMMITMENT; + } Assume(inv_type != 0); // guarded by the early-return above auto pm = std::make_shared(std::move(vRecv)); - CHashWriter hw(SER_GETHASH, 0); - hw.write(AsWritableBytes(Span{*pm})); - const uint256 hash = hw.GetHash(); - const NodeId from = pfrom.GetId(); const bool dispatched = m_qdkgsman.DoForHandler({llmqType, quorumIndex}, [&](CDKGSessionHandler& handler) { CDKGPendingMessages* pending = nullptr; @@ -679,6 +862,9 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) handler.WaitForNextPhase(std::nullopt, QuorumPhase::Initialized); + // Leftovers missed their matching phase and cannot be accepted by this + // round. Discard their raw bytes without materializing BLS objects on the + // critical path to initializing the next quorum. handler.ClearPendingMessages(); uint256 curQuorumHash = handler.GetCurrentQuorumHash(); @@ -724,8 +910,8 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) } }; auto fContributeWait = [this, curSession, &handler, &active] { - return ProcessPendingMessageBatch(active.connman, *curSession, handler.pendingContributions, - *m_peer_manager, 8); + return ProcessPendingMessageBatch(active.connman, *curSession, handler.params, + handler.pendingContributions, *m_peer_manager, 8); }; handler.HandlePhase(QuorumPhase::Contribute, QuorumPhase::Complain, curQuorumHash, 0.05, fContributeStart, fContributeWait); @@ -737,8 +923,8 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) } }; auto fComplainWait = [this, curSession, &handler, &active] { - return ProcessPendingMessageBatch(active.connman, *curSession, handler.pendingComplaints, - *m_peer_manager, 8); + return ProcessPendingMessageBatch(active.connman, *curSession, handler.params, + handler.pendingComplaints, *m_peer_manager, 8); }; handler.HandlePhase(QuorumPhase::Complain, QuorumPhase::Justify, curQuorumHash, 0.05, fComplainStart, fComplainWait); @@ -749,8 +935,8 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) } }; auto fJustifyWait = [this, curSession, &handler, &active] { - return ProcessPendingMessageBatch(active.connman, *curSession, handler.pendingJustifications, - *m_peer_manager, 8); + return ProcessPendingMessageBatch(active.connman, *curSession, handler.params, + handler.pendingJustifications, *m_peer_manager, 8); }; handler.HandlePhase(QuorumPhase::Justify, QuorumPhase::Commit, curQuorumHash, 0.05, fJustifyStart, fJustifyWait); @@ -761,7 +947,7 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) } }; auto fCommitWait = [this, curSession, &handler, &active] { - return ProcessPendingMessageBatch(active.connman, *curSession, + return ProcessPendingMessageBatch(active.connman, *curSession, handler.params, handler.pendingPrematureCommitments, *m_peer_manager, 8); }; diff --git a/src/test/llmq_dkg_tests.cpp b/src/test/llmq_dkg_tests.cpp index 9715a63f2719..bd34ce59672d 100644 --- a/src/test/llmq_dkg_tests.cpp +++ b/src/test/llmq_dkg_tests.cpp @@ -3,6 +3,9 @@ // file COPYING or http://www.opensource.org/licenses/mit-license.php. #include +#include +#include +#include #include #include @@ -23,4 +26,69 @@ BOOST_AUTO_TEST_CASE(llmq_dkgerror) BOOST_REQUIRE(GetSimulatedErrorRate(llmq::DKGError::type::_COUNT) == 0.0); } +BOOST_AUTO_TEST_CASE(pending_messages_local_first_uses_full_remote_allowance) +{ + using namespace llmq; + + auto make_message = [] { return std::make_shared(SER_NETWORK, PROTOCOL_VERSION); }; + auto make_hash = [](uint8_t value) { + uint256 hash; + hash.begin()[0] = value; + return hash; + }; + + CDKGPendingMessages pending{/*max_messages_per_node=*/2}; + pending.PushPendingMessage(/*from=*/-1, make_message(), make_hash(1)); + pending.PushPendingMessage(/*from=*/1, make_message(), make_hash(2)); + pending.PushPendingMessage(/*from=*/1, make_message(), make_hash(3)); + pending.PushPendingMessage(/*from=*/2, make_message(), make_hash(4)); + pending.PushPendingMessage(/*from=*/2, make_message(), make_hash(5)); + + BOOST_CHECK_EQUAL(pending.PopPendingMessages(6).size(), 5U); +} + +BOOST_AUTO_TEST_CASE(pending_messages_bounded_across_node_ids) +{ + using namespace llmq; + + auto make_message = [] { return std::make_shared(SER_NETWORK, PROTOCOL_VERSION); }; + auto make_hash = [](uint8_t value) { + uint256 hash; + hash.begin()[0] = value; + return hash; + }; + + CDKGPendingMessages pending{/*max_messages_per_node=*/2}; + pending.PushPendingMessage(/*from=*/1, make_message(), make_hash(1)); + pending.PushPendingMessage(/*from=*/1, make_message(), make_hash(2)); + + // One peer's full quota does not consume the queue-wide remote allowance. + pending.PushPendingMessage(/*from=*/2, make_message(), make_hash(3)); + pending.PushPendingMessage(/*from=*/2, make_message(), make_hash(4)); + BOOST_CHECK(pending.HasSeen(make_hash(3))); + BOOST_CHECK(pending.HasSeen(make_hash(4))); + + // Fresh NodeIds cannot bypass the queue-wide remote-message cap. + pending.PushPendingMessage(/*from=*/3, make_message(), make_hash(5)); + pending.PushPendingMessage(/*from=*/4, make_message(), make_hash(6)); + BOOST_CHECK(!pending.HasSeen(make_hash(5))); + BOOST_CHECK(!pending.HasSeen(make_hash(6))); + + // Duplicates are rejected before charging the new NodeId's quota. + pending.PushPendingMessage(/*from=*/3, make_message(), make_hash(1)); + pending.PushPendingMessage(/*from=*/3, make_message(), make_hash(1)); + + BOOST_CHECK_EQUAL(pending.PopPendingMessages(5).size(), 4U); + pending.PushPendingMessage(/*from=*/3, make_message(), make_hash(7)); + pending.PushPendingMessage(/*from=*/3, make_message(), make_hash(8)); + pending.PushPendingMessage(/*from=*/3, make_message(), make_hash(9)); + BOOST_CHECK(pending.HasSeen(make_hash(7))); + BOOST_CHECK(pending.HasSeen(make_hash(8))); + BOOST_CHECK(!pending.HasSeen(make_hash(9))); + + pending.Clear(); + pending.PushPendingMessage(/*from=*/3, make_message(), make_hash(7)); + BOOST_CHECK(pending.HasSeen(make_hash(7))); +} + BOOST_AUTO_TEST_SUITE_END() diff --git a/test/functional/feature_llmq_dkg_intake.py b/test/functional/feature_llmq_dkg_intake.py index d626d178a463..c953e05f4dd4 100755 --- a/test/functional/feature_llmq_dkg_intake.py +++ b/test/functional/feature_llmq_dkg_intake.py @@ -10,10 +10,15 @@ not MNAuth-verified are rejected before retention. - oversized DKG payloads are rejected (before deserialization / retention) even from a verified peer. - - structural pre-validation: malformed DKG payloads (valid quorum prefix, garbage - body) are rejected before retention even from a verified peer. + - structural pre-validation: truncated, trailing, or parametrically out-of-bounds + DKG payloads are rejected before retention even from a verified peer. + - BLS objects are not materialized at intake: a structurally plausible payload with + an invalid BLS encoding reaches the DKG worker and is rejected there. + - late, framing-valid messages are bounded across reconnect-generated NodeIds and + discarded without BLS materialization before the next round initializes. -The node must not crash; the sending peer must be scored (Misbehaving). +The node must not crash; rejected malformed messages must be scored where the +matching worker still processes them. """ from test_framework.messages import ser_compact_size, ser_uint256 @@ -29,6 +34,8 @@ FAKE_PUBKEY = "8e7afdb849e5e2a085b035b62e21c0940c753f2d4501325743894c37162f287bccaffbedd60c36581dabbf127a22e43f" DKG_PUSH_TYPES = [b"qcontrib", b"qcomplaint", b"qjustify", b"qpcommit"] +VALID_BLS_PUBKEY = bytes.fromhex(FAKE_PUBKEY) +INVALID_NONZERO_BLS_PUBKEY = b"\xff" * 48 class msg_dkg_raw: @@ -46,10 +53,12 @@ def __repr__(self): return "msg_dkg_raw(type=%s, len=%d)" % (self.msgtype, len(self.payload)) -def get_p2p_id(node): +def get_p2p_id(node, uacomment=None): def get_id(): for p in node.getpeerinfo(): for p2p in node.p2ps: + if uacomment is not None and p2p.uacomment != uacomment: + continue if p["subver"] == p2p.strSubVer: return p["id"] return None @@ -73,8 +82,14 @@ def add_options(self, parser): def set_test_params(self): # -whitelist keeps the adversarial peer connected even after it crosses the # discouragement threshold, so banscore stays observable for the score==100 cases. - # -debug=net surfaces the Misbehaving reason strings in debug.log. - extra_args = [["-whitelist=127.0.0.1", "-debug=net", "-deprecatedrpc=banscore"]] * 4 + # -debug=net surfaces the Misbehaving reason strings in debug.log, while + # -debug=llmq-dkg exposes worker and queue-boundary behavior. + extra_args = [[ + "-whitelist=127.0.0.1", + "-debug=net", + "-debug=llmq-dkg", + "-deprecatedrpc=banscore", + ]] * 4 self.set_dash_test_params(4, 3, extra_args=extra_args) def quorum_hash_prefix(self): @@ -83,14 +98,14 @@ def quorum_hash_prefix(self): # real in-progress quorum and reach the size/structural checks. return bytes([LLMQ_TEST]) + ser_uint256(int(self.quorum_hash, 16)) - def qcontrib_payload(self, blob_count): + def qcontrib_payload(self, blob_count, vvec_pubkey=VALID_BLS_PUBKEY, protx_hash=0): # CDKGContribution: llmqType, quorumHash, proTxHash, vvec, contributions, sig. # LLMQ_TEST uses threshold=2/minSize=2 by default, so blob_count=1 is # well-formed enough to deserialize but below the contribution lower bound. r = self.quorum_hash_prefix() - r += ser_uint256(0) # proTxHash - r += ser_compact_size(2) + b"\x00" * (2 * 48) # BLSVerificationVector - r += b"\x00" * 48 # CBLSIESMultiRecipientBlobs::ephemeralPubKey + r += ser_uint256(protx_hash) + r += ser_compact_size(2) + vvec_pubkey + VALID_BLS_PUBKEY # BLSVerificationVector + r += VALID_BLS_PUBKEY # CBLSIESMultiRecipientBlobs::ephemeralPubKey r += b"\x00" * 32 # CBLSIESMultiRecipientBlobs::ivSeed r += ser_compact_size(blob_count) for _ in range(blob_count): @@ -98,9 +113,9 @@ def qcontrib_payload(self, blob_count): r += b"\x00" * 96 # sig return r - def add_verified_peer(self, node): - peer = node.add_p2p_connection(P2PInterface()) - peer_id = get_p2p_id(node) + def add_verified_peer(self, node, uacomment=None): + peer = node.add_p2p_connection(P2PInterface(), uacomment=uacomment) + peer_id = get_p2p_id(node, uacomment) assert node.mnauth(peer_id, FAKE_PROTX, FAKE_PUBKEY) return peer, peer_id @@ -118,6 +133,9 @@ def run_test(self): self.test_unverified_sender_rejected(mn_node) self.test_oversized_rejected(mn_node) self.test_malformed_rejected(mn_node) + self.test_trailing_bytes_rejected(mn_node) + self.test_malformed_bls_pubkey_rejected_by_worker(mn_node) + self.test_late_messages_bounded_across_reconnects(mn_node) self.test_under_min_contribution_blobs_rejected(mn_node) def test_unverified_sender_rejected(self, node): @@ -160,6 +178,112 @@ def test_malformed_rejected(self, node): wait_for_banscore(node, peer_id, 100) node.disconnect_p2ps() + def test_trailing_bytes_rejected(self, node): + self.log.info("QCONTRIB with trailing bytes is rejected at intake (Misbehaving 100)") + peer, peer_id = self.add_verified_peer(node) + wait_for_banscore(node, peer_id, 0) + with node.assert_debug_log(["malformed DKG message"]): + peer.send_message(msg_dkg_raw( + b"qcontrib", + self.qcontrib_payload(blob_count=2) + b"\x00", + )) + peer.sync_with_ping() + wait_for_banscore(node, peer_id, 100) + node.disconnect_p2ps() + + def _start_fresh_dkg_cycle(self, nodes): + """Land on the base block of a fresh DKG cycle (phase 1 / Initialized).""" + cycle_length = 24 + skip_count = cycle_length - (self.nodes[0].getblockcount() % cycle_length) + self.generate(self.nodes[0], skip_count, sync_fun=lambda: self.sync_blocks(nodes)) + self.quorum_hash = self.nodes[0].getbestblockhash() + self.wait_for_quorum_phase(self.quorum_hash, 1, self.llmq_size, None, 0, self.mninfo) + + def test_malformed_bls_pubkey_rejected_by_worker(self, node): + self.log.info("QCONTRIB BLS decoding is deferred to the DKG worker (Misbehaving 100)") + nodes = [self.nodes[0]] + [mn.get_node(self) for mn in self.mninfo] + # Queue during Initialized; Contribute's matching drain deserializes and scores. + self._start_fresh_dkg_cycle(nodes) + + peer, peer_id = self.add_verified_peer(node) + wait_for_banscore(node, peer_id, 0) + with node.assert_debug_log( + ["failed to deserialize message"], + unexpected_msgs=["malformed DKG message"], + timeout=10, + ): + peer.send_message(msg_dkg_raw(b"qcontrib", self.qcontrib_payload( + blob_count=2, + vvec_pubkey=INVALID_NONZERO_BLS_PUBKEY, + ))) + peer.sync_with_ping() + wait_for_banscore(node, peer_id, 0) + self.move_blocks(nodes, 2) + wait_for_banscore(node, peer_id, 100) + node.disconnect_p2ps() + + def test_late_messages_bounded_across_reconnects(self, node): + self.log.info("Late QCONTRIB retention is bounded across reconnects and cleared without BLS decoding") + nodes = [self.nodes[0]] + [mn.get_node(self) for mn in self.mninfo] + cycle_length = 24 + self._start_fresh_dkg_cycle(nodes) + stage = self.nodes[0].getblockcount() % cycle_length + assert stage == 0, "expected DKG cycle base, got stage %d" % stage + # phaseBlocks=2: stage 0=Initialized, 2=Contribute, 4=Complain. + complain_stage = 4 + self.move_blocks(nodes, complain_stage - stage) + assert self.nodes[0].getblockcount() % cycle_length == complain_stage + + # Each transient connection gets a fresh NodeId. Unique proTxHash bytes + # avoid deduplication, so this specifically exercises the queue-wide cap + # rather than the per-NodeId quota. Keep the final accepted peer connected + # to verify that round-start clearing does not score stale BLS encodings. + queue_limit = 4 * self.llmq_size + retained_peer = None + retained_peer_id = None + for nonce in range(1, queue_limit + 1): + uacomment = "dkg-late-%d" % nonce + peer, peer_id = self.add_verified_peer(node, uacomment) + peer.send_message(msg_dkg_raw(b"qcontrib", self.qcontrib_payload( + blob_count=2, + vvec_pubkey=INVALID_NONZERO_BLS_PUBKEY, + protx_hash=nonce, + ))) + peer.sync_with_ping() + wait_for_banscore(node, peer_id, 0) + if nonce == queue_limit: + retained_peer = peer + retained_peer_id = peer_id + else: + peer.peer_disconnect() + peer.wait_for_disconnect() + + overflow_peer, overflow_peer_id = self.add_verified_peer(node, "dkg-late-overflow") + with node.assert_debug_log(["pending queue full"]): + overflow_peer.send_message(msg_dkg_raw(b"qcontrib", self.qcontrib_payload( + blob_count=2, + vvec_pubkey=INVALID_NONZERO_BLS_PUBKEY, + protx_hash=queue_limit + 1, + ))) + overflow_peer.sync_with_ping() + wait_for_banscore(node, overflow_peer_id, 0) + + # Crossing the phase boundary must clear a bounded raw queue and finish + # initializing the next session without deserializing stale BLS points. + remaining = cycle_length - (self.nodes[0].getblockcount() % cycle_length) + with node.assert_debug_log( + [], + unexpected_msgs=["malformed DKG message", "failed to deserialize message"], + timeout=60, + ): + self.move_blocks(nodes, remaining) + self.quorum_hash = self.nodes[0].getbestblockhash() + self.wait_for_quorum_phase(self.quorum_hash, 1, self.llmq_size, None, 0, self.mninfo) + assert retained_peer is not None + wait_for_banscore(node, retained_peer_id, 0) + wait_for_banscore(node, overflow_peer_id, 0) + node.disconnect_p2ps() + def test_under_min_contribution_blobs_rejected(self, node): self.log.info("QCONTRIB with fewer than minSize encrypted blobs is rejected (Misbehaving 100)") peer, peer_id = self.add_verified_peer(node)