diff --git a/src/llmq/dkgsessionhandler.cpp b/src/llmq/dkgsessionhandler.cpp index c9258ff353f9..bd06432851c9 100644 --- a/src/llmq/dkgsessionhandler.cpp +++ b/src/llmq/dkgsessionhandler.cpp @@ -29,6 +29,13 @@ void CDKGPendingMessages::PushPendingMessage(NodeId from, std::shared_ptr= maxMessagesPerNode) { // TODO ban? LogPrint(BCLog::LLMQ_DKG, "CDKGPendingMessages::%s -- too many messages, peer=%d\n", __func__, from); @@ -36,11 +43,7 @@ void CDKGPendingMessages::PushPendingMessage(NodeId from, std::shared_ptr #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 { @@ -63,7 +61,7 @@ class CDKGPendingMessages public: explicit CDKGPendingMessages(size_t _maxMessagesPerNode) : - maxMessagesPerNode(_maxMessagesPerNode) {}; + maxMessagesPerNode(_maxMessagesPerNode) {} /** * Enqueue a serialized DKG message under @p from with content hash @p hash. @@ -77,31 +75,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..66b10290792f 100644 --- a/src/llmq/net_dkg.cpp +++ b/src/llmq/net_dkg.cpp @@ -71,46 +71,209 @@ 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); +} + +template +bool ReadAndCheckBLSObject(CDataStream& ds) +{ + BLSObject obj; + try { + ds >> obj; + } catch (const std::exception&) { + return false; + } + return obj.IsValid(); +} + +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 (!ReadAndCheckBLSObject(ds)) { + return false; } + } + + if (!ReadAndCheckBLSObject(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 ReadAndCheckBLSObject(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 && + ReadAndCheckBLSObject(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) || !ReadAndCheckBLSObject(ds)) { + return false; + } + } + + return ReadAndCheckBLSObject(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) && + ReadAndCheckBLSObject(ds) && + SkipBytes(ds, 32) && + ReadAndCheckBLSObject(ds) && + ReadAndCheckBLSObject(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 +420,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 +433,32 @@ 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 +468,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 +483,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; @@ -403,10 +593,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 +654,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; @@ -724,7 +921,7 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) } }; auto fContributeWait = [this, curSession, &handler, &active] { - return ProcessPendingMessageBatch(active.connman, *curSession, handler.pendingContributions, + return ProcessPendingMessageBatch(active.connman, *curSession, handler.params, handler.pendingContributions, *m_peer_manager, 8); }; handler.HandlePhase(QuorumPhase::Contribute, QuorumPhase::Complain, curQuorumHash, 0.05, fContributeStart, @@ -737,7 +934,7 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) } }; auto fComplainWait = [this, curSession, &handler, &active] { - return ProcessPendingMessageBatch(active.connman, *curSession, handler.pendingComplaints, + 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,7 +946,7 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) } }; auto fJustifyWait = [this, curSession, &handler, &active] { - return ProcessPendingMessageBatch(active.connman, *curSession, handler.pendingJustifications, + return ProcessPendingMessageBatch(active.connman, *curSession, handler.params, handler.pendingJustifications, *m_peer_manager, 8); }; handler.HandlePhase(QuorumPhase::Justify, QuorumPhase::Commit, curQuorumHash, 0.05, fJustifyStart, fJustifyWait); @@ -762,8 +959,8 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) }; auto fCommitWait = [this, curSession, &handler, &active] { return ProcessPendingMessageBatch(active.connman, *curSession, - handler.pendingPrematureCommitments, *m_peer_manager, - 8); + handler.params, handler.pendingPrematureCommitments, + *m_peer_manager, 8); }; handler.HandlePhase(QuorumPhase::Commit, QuorumPhase::Finalize, curQuorumHash, 0.1, fCommitStart, fCommitWait); diff --git a/test/functional/feature_llmq_dkg_intake.py b/test/functional/feature_llmq_dkg_intake.py index d626d178a463..8c3009bd92c6 100755 --- a/test/functional/feature_llmq_dkg_intake.py +++ b/test/functional/feature_llmq_dkg_intake.py @@ -12,6 +12,8 @@ from a verified peer. - structural pre-validation: malformed DKG payloads (valid quorum prefix, garbage body) are rejected before retention even from a verified peer. + - BLS pre-validation: malformed non-zero BLS encodings are rejected before + retention even when the payload's lengths and counts are structurally plausible. The node must not crash; the sending peer must be scored (Misbehaving). """ @@ -29,6 +31,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: @@ -83,14 +87,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): # 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_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): @@ -118,6 +122,7 @@ 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_malformed_bls_pubkey_rejected(mn_node) self.test_under_min_contribution_blobs_rejected(mn_node) def test_unverified_sender_rejected(self, node): @@ -160,6 +165,19 @@ def test_malformed_rejected(self, node): wait_for_banscore(node, peer_id, 100) node.disconnect_p2ps() + def test_malformed_bls_pubkey_rejected(self, node): + self.log.info("QCONTRIB with a malformed non-zero BLS pubkey is rejected (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, + vvec_pubkey=INVALID_NONZERO_BLS_PUBKEY, + ))) + peer.sync_with_ping() + wait_for_banscore(node, peer_id, 100) + 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)