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)