From 2aefa6ee3836c87601874e9d157e4607746081db Mon Sep 17 00:00:00 2001 From: pasta Date: Fri, 7 Aug 2026 15:43:02 -0500 Subject: [PATCH 1/3] perf(llmq): deserialize DKG messages once DKG intake previously deserialized a copy of each accepted payload (repeating BLS point decompression on the shared network thread) for structural validation, and the DKG worker then deserialized the retained bytes again. Replace the typed intake pass with a framing-only wire walk that validates CompactSize counts, dynamic bitsets (via the same ReadFixedBitSet the typed path uses), quorum-parameter bounds, truncation, and trailing bytes without decoding any BLS object. The worker is now the sole typed deserialization point, immediately followed by the same parameter-derived structural checks. The pre-existing per-peer pending-message quota is rekeyed from NodeId to the MNAuth-verified proTxHash and made cumulative for the round, so a sender can no longer reset its retention budget by reconnecting or by waiting for the worker to drain the queue. Own messages are enqueued under this node's own proTxHash and share the same quota path. Sender identities are pinned to the deterministic masternode list by MNAuth, so worst-case retention is bounded by (hostile MN count) x quota. Duplicate hashes are rejected before charging the quota, and quota-dropped messages are not marked seen so another peer with budget can re-deliver them. The llmqType/quorumHash prefix is peeked via SpanReader instead of read+Rewind, and short payloads are scored instead of throwing out of ProcessMessage. Leftover raw queues are discarded at round start without BLS work. Co-Authored-By: Claude Fable 5 --- src/llmq/dkgsessionhandler.h | 25 ---- src/llmq/net_dkg.cpp | 240 +++++++++++++++++++++++++++-------- src/llmq/net_dkg.h | 39 +++++- 3 files changed, 221 insertions(+), 83 deletions(-) diff --git a/src/llmq/dkgsessionhandler.h b/src/llmq/dkgsessionhandler.h index 57e5ecaef322..f52bbc81c416 100644 --- a/src/llmq/dkgsessionhandler.h +++ b/src/llmq/dkgsessionhandler.h @@ -83,31 +83,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 b42b0fdf021e..9c6678262abf 100644 --- a/src/llmq/net_dkg.cpp +++ b/src/llmq/net_dkg.cpp @@ -22,7 +22,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -72,48 +74,144 @@ 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) +constexpr size_t DKG_MSG_PREFIX_SIZE = 1 + 32 + 32; // llmqType + quorumHash + proTxHash + +// The framing walks below use throwing stream primitives; truncation anywhere +// surfaces as an exception the dispatcher turns into a reject. They return +// false on a params bound violation and true only when the whole payload was +// consumed. BLS encodings have fixed wire sizes, so the walks only establish +// that the bytes are present; decoding happens once, on the DKG worker. + +// Reads a DYNBITSET bounded by @p max_size and returns its bit count. Reuses +// ReadFixedBitSet -- the exact deserializer the typed path uses -- so +// truncation and padding-bit handling cannot diverge. +uint64_t ReadBoundedDynBitset(CDataStream& ds, size_t max_size) +{ + const uint64_t bit_count = ReadCompactSize(ds); + if (bit_count > max_size) { + throw std::ios_base::failure("dynamic bitset exceeds quorum size"); + } + std::vector bits; + ReadFixedBitSet(ds, bits, bit_count); + return bit_count; +} + +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; + + ds.ignore(DKG_MSG_PREFIX_SIZE); + if (ReadCompactSize(ds) != threshold) { // vvec + return false; + } + ds.ignore(threshold * CBLSPublicKey::SerSize); + ds.ignore(CBLSPublicKey::SerSize + 32); // IES ephemeralPubKey + ivSeed + const uint64_t blob_count = ReadCompactSize(ds); + if (blob_count < min_size || blob_count > size) { + return false; + } + for (uint64_t i = 0; i < blob_count; ++i) { + ds.ignore(ReadCompactSize(ds)); // encrypted contribution blob + } + ds.ignore(CBLSSignature::SerSize); + return ds.empty(); +} + +bool CheckComplaintWireStructure(CDataStream& ds, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + + ds.ignore(DKG_MSG_PREFIX_SIZE); + const uint64_t bad_members = ReadBoundedDynBitset(ds, size); + const uint64_t complain_for_members = ReadBoundedDynBitset(ds, size); + if (bad_members != complain_for_members) { + return false; + } + ds.ignore(CBLSSignature::SerSize); + return ds.empty(); +} + +bool CheckJustificationWireStructure(CDataStream& ds, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + + ds.ignore(DKG_MSG_PREFIX_SIZE); + const uint64_t contribution_count = ReadCompactSize(ds); + if (contribution_count > size) { + return false; + } + ds.ignore(contribution_count * (4 + CBLSSecretKey::SerSize)); // {u32 index, encrypted key share} + ds.ignore(CBLSSignature::SerSize); + return ds.empty(); +} + +bool CheckPrematureCommitmentWireStructure(CDataStream& ds, const Consensus::LLMQParams& params) +{ + const size_t size = params.size > 0 ? static_cast(params.size) : 0; + + ds.ignore(DKG_MSG_PREFIX_SIZE); + ReadBoundedDynBitset(ds, size); // validMembers + // quorumPublicKey + quorumVvecHash + quorumSig + sig + ds.ignore(CBLSPublicKey::SerSize + 32 + 2 * CBLSSignature::SerSize); + return ds.empty(); +} + +} // namespace + +bool CheckDKGMessageWireStructure(std::string_view msg_type, const CDataStream& payload, + const Consensus::LLMQParams& params) +{ + // Walk a copy so the caller's read position (and the bytes backing its + // inventory hash) survive. The copy is a memcpy bounded by the size cap + // applied just before this call -- negligible next to the BLS point + // decompression that deserializing the payload here would have cost. + CDataStream ds{payload}; 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; + return CheckContributionWireStructure(ds, params); } else if (msg_type == NetMsgType::QCOMPLAINT) { - CDKGComplaint qc; - s >> qc; - return qc.badMembers.size() == qc.complainForMembers.size() && - qc.badMembers.size() <= size; + return CheckComplaintWireStructure(ds, params); } else if (msg_type == NetMsgType::QJUSTIFICATION) { - CDKGJustification qj; - s >> qj; - return qj.contributions.size() <= size; + return CheckJustificationWireStructure(ds, params); } else if (msg_type == NetMsgType::QPCOMMITMENT) { - CDKGPrematureCommitment qc; - s >> qc; - return qc.validMembers.size() <= size; + return CheckPrematureCommitmentWireStructure(ds, params); } - return false; } catch (const std::exception&) { - return false; } + return false; +} + +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; +} + +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; +} + +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; +} + +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; } +namespace { + // returns a set of NodeIds which sent invalid messages template std::unordered_set BatchVerifyMessageSigs(CDKGSession& session, @@ -266,11 +364,35 @@ void EnqueueOwn(CDKGPendingMessages& pending, const uint256& own_protx, const Me pending.PushPendingMessage(/*from=*/-1, own_protx, std::move(pm), hw.GetHash()); } +// The single typed deserialization pass over a queued payload, followed by the +// param-derived structural bounds. Returns nullptr on failure; caller scores. +template +std::shared_ptr DeserializeCheckedDKGMessage(CDataStream& ds, const Consensus::LLMQParams& params, + NodeId nodeId) +{ + auto msg = std::make_shared(); + try { + ds >> *msg; + } catch (...) { + LogPrint(BCLog::LLMQ_DKG, "%s -- failed to deserialize message, peer=%d\n", __func__, nodeId); + return nullptr; + } + if (!ds.empty()) { + LogPrint(BCLog::LLMQ_DKG, "%s -- failed to deserialize message, peer=%d\n", __func__, nodeId); + return nullptr; + } + if (!CheckDKGMessageStructure(*msg, params)) { + LogPrint(BCLog::LLMQ_DKG, "%s -- message failed structure check, peer=%d\n", __func__, nodeId); + return nullptr; + } + return msg; +} + template -bool ProcessPendingMessageBatch(const CConnman& connman, CDKGSession& session, CDKGPendingMessages& pendingMessages, - PeerManagerInternal& peerman, size_t maxCount) +bool ProcessPendingMessageBatch(const CConnman& connman, CDKGSession& session, const Consensus::LLMQParams& params, + CDKGPendingMessages& pendingMessages, PeerManagerInternal& peerman, size_t maxCount) { - auto msgs = pendingMessages.PopAndDeserializeMessages(maxCount); + auto msgs = pendingMessages.PopPendingMessages(maxCount); if (msgs.empty()) { return false; } @@ -280,13 +402,13 @@ bool ProcessPendingMessageBatch(const CConnman& connman, CDKGSession& session, C for (const auto& p : msgs) { const NodeId& nodeId = p.first; - if (!p.second) { - LogPrint(BCLog::LLMQ_DKG, "%s -- failed to deserialize message, peer=%d\n", __func__, nodeId); + auto msg = DeserializeCheckedDKGMessage(*p.second, params, nodeId); + if (!msg) { 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); @@ -294,7 +416,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; @@ -376,10 +498,8 @@ void NetDKG::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStre return; } - const bool is_masternode = m_active != nullptr; - if (msg_type == NetMsgType::QWATCH) { - if (!is_masternode) { + if (m_active == nullptr) { // non-masternodes should never receive this m_peer_manager->PeerMisbehaving(pfrom.GetId(), 10); return; @@ -388,6 +508,15 @@ void NetDKG::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStre return; } + // Observer handlers have no DKG worker and therefore cannot validate or + // drain queued payloads. Observers learn completed quorums through the + // quorum-data path; retaining round messages here would only create + // process-lifetime state. + if (m_active == nullptr) { + LogPrint(BCLog::LLMQ_DKG, "NetDKG -- ignoring %s in observer mode\n", msg_type); + return; + } + // Pushed DKG messages (QCONTRIB/QCOMPLAINT/QJUSTIFICATION/QPCOMMITMENT) retain // attacker-controlled payloads, so they must originate from an MNAuth-verified // masternode. qwatch is unauthenticated (any peer can set it via QWATCH) and is @@ -399,17 +528,16 @@ void NetDKG::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStre return; } - if (vRecv.empty()) { - m_peer_manager->PeerMisbehaving(pfrom.GetId(), 100); + if (vRecv.size() < sizeof(uint8_t) + sizeof(uint256)) { + m_peer_manager->PeerMisbehaving(pfrom.GetId(), 100, "malformed DKG message"); return; } + // Peek the prefix without touching vRecv's read position; the full payload + // stays available for the framing walk, inventory hash, and retention below Consensus::LLMQType llmqType; uint256 quorumHash; - vRecv >> llmqType; - vRecv >> quorumHash; - vRecv.Rewind(sizeof(uint256)); - vRecv.Rewind(sizeof(uint8_t)); + SpanReader{vRecv.GetType(), vRecv.GetVersion(), MakeUCharSpan(vRecv)} >> llmqType >> quorumHash; const auto& llmq_params_opt = Params().GetLLMQ(llmqType); if (!llmq_params_opt.has_value()) { @@ -467,9 +595,9 @@ void NetDKG::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStre 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)) { + // Framing-only pre-validation before retention; BLS decoding happens + // exactly once, on the DKG worker, after retention. + if (!CheckDKGMessageWireStructure(msg_type, vRecv, llmq_params)) { m_peer_manager->PeerMisbehaving(pfrom.GetId(), 100, "malformed DKG message"); return; } @@ -544,6 +672,9 @@ bool NetDKG::AlreadyHave(const CInv& inv) case MSG_QUORUM_COMPLAINT: case MSG_QUORUM_JUSTIFICATION: case MSG_QUORUM_PREMATURE_COMMITMENT: { + // Observers have no worker that could consume or validate DKG round + // payloads, so do not request them in response to inventory. + if (m_active == nullptr) return true; if (!IsQuorumDKGEnabled(m_sporkman)) return false; bool seen = false; m_qdkgsman.ForEachHandler([&](const CDKGSessionHandler& h) { @@ -705,6 +836,7 @@ void NetDKG::HandleDKGRound(ActiveDKGSessionHandler& handler) handler.WaitForNextPhase(std::nullopt, QuorumPhase::Initialized); + // Leftovers missed their phase; discard raw bytes without BLS work handler.ClearPendingMessages(); uint256 curQuorumHash = handler.GetCurrentQuorumHash(); @@ -750,8 +882,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); @@ -763,8 +895,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); @@ -775,8 +907,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); @@ -787,7 +919,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/llmq/net_dkg.h b/src/llmq/net_dkg.h index 59ea5882c4e8..f5e72477075b 100644 --- a/src/llmq/net_dkg.h +++ b/src/llmq/net_dkg.h @@ -13,6 +13,7 @@ #include #include +#include #include #include @@ -25,7 +26,11 @@ class CMasternodeMetaMan; class CSporkManager; namespace llmq { class ActiveDKGSessionHandler; +class CDKGComplaint; +class CDKGContribution; class CDKGDebugManager; +class CDKGJustification; +class CDKGPrematureCommitment; class CDKGSessionManager; class CQuorumBlockProcessor; class CQuorumManager; @@ -34,17 +39,43 @@ class QuorumRole; } // namespace llmq namespace llmq { + +/** + * Framing-only validation of a raw DKG payload, run at network intake before + * retention: checks truncation, trailing bytes, and quorum-param bounds on a + * copy of @p payload without decoding any BLS object. Typed deserialization + * happens exactly once, later, on the DKG worker thread. + * + * @warning This walk mirrors the (Un)serialize implementations in + * llmq/dkgmessages.h. Any change to those must be reflected here; + * src/test/fuzz/dkg_message_framing.cpp asserts that this never + * rejects a payload the worker would accept. + */ +bool CheckDKGMessageWireStructure(std::string_view msg_type, const CDataStream& payload, + const Consensus::LLMQParams& params); + +/** + * Param-only structural bounds on a typed DKG message, run by the DKG worker + * right after deserializing queued bytes and before PreVerifyMessage. + */ +bool CheckDKGMessageStructure(const CDKGContribution& qc, const Consensus::LLMQParams& params); +bool CheckDKGMessageStructure(const CDKGComplaint& qc, const Consensus::LLMQParams& params); +bool CheckDKGMessageStructure(const CDKGJustification& qj, const Consensus::LLMQParams& params); +bool CheckDKGMessageStructure(const CDKGPrematureCommitment& qc, const Consensus::LLMQParams& params); + /** * NetHandler responsible for DKG networking: * - QCONTRIB / QCOMPLAINT / QJUSTIFICATION / QPCOMMITMENT / QWATCH ProcessMessage - * routing into CDKGSessionManager. The resulting MessageProcessingResult is - * consumed locally via PeerManagerInternal and never propagated up. - * - AlreadyHave for the four MSG_QUORUM_* DKG inv types. + * routing into CDKGSessionManager in active mode. Observer mode ignores DKG + * round payloads because it has no worker to consume them. + * - AlreadyHave for the four MSG_QUORUM_* DKG inv types. Observer mode returns + * true so it does not request payloads it cannot consume. * - ProcessGetData for the four MSG_QUORUM_* DKG inv types (active mode only; * in observer mode the underlying Get* calls return false by construction). * * Active-mode-only deps live in @ref ActiveDKG; @ref m_active is null in - * observer mode and non-null in active mode (all-or-none). + * observer mode and non-null in active mode (all-or-none). Observers neither + * request nor retain DKG round payloads. * * On nodes that run neither active nor observer mode, register @ref NetDKGStub * instead. From 90bf7257b722df92e044273132fbfe34322eb54f Mon Sep 17 00:00:00 2001 From: pasta Date: Fri, 7 Aug 2026 15:43:27 -0500 Subject: [PATCH 2/3] test: add dkg_message_framing fuzz target Intake framing validation and worker typed deserialization are two hand-maintained parsers over one wire format. The safety-critical direction is that framing must never reject a payload the worker would accept, otherwise honest DKG messages are silently dropped before retention and quorum formation degrades. Assert that direction over fuzzer-provided payloads for every configured LLMQ and both BLS schemes, plus a constructed well-formed message per input so serializer/framing drift is caught even from an empty corpus. The converse is intentionally not asserted: framing accepts undecodable BLS encodings so the worker can score the sender. Co-Authored-By: Claude Fable 5 --- src/Makefile.test.include | 1 + src/test/fuzz/dkg_message_framing.cpp | 157 ++++++++++++++++++++++++++ test/util/data/non-backported.txt | 1 + 3 files changed, 159 insertions(+) create mode 100644 src/test/fuzz/dkg_message_framing.cpp diff --git a/src/Makefile.test.include b/src/Makefile.test.include index 427909b8e12f..0cbd534132a4 100644 --- a/src/Makefile.test.include +++ b/src/Makefile.test.include @@ -321,6 +321,7 @@ test_fuzz_fuzz_SOURCES = \ test/fuzz/decode_tx.cpp \ test/fuzz/descriptor_parse.cpp \ test/fuzz/deserialize.cpp \ + test/fuzz/dkg_message_framing.cpp \ test/fuzz/eval_script.cpp \ test/fuzz/fee_rate.cpp \ test/fuzz/fees.cpp \ diff --git a/src/test/fuzz/dkg_message_framing.cpp b/src/test/fuzz/dkg_message_framing.cpp new file mode 100644 index 000000000000..1c300b176b2a --- /dev/null +++ b/src/test/fuzz/dkg_message_framing.cpp @@ -0,0 +1,157 @@ +// Copyright (c) 2025 The Dash Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include + +namespace { +const std::array DKG_MSG_TYPES = { + NetMsgType::QCONTRIB, + NetMsgType::QCOMPLAINT, + NetMsgType::QJUSTIFICATION, + NetMsgType::QPCOMMITMENT, +}; + +//! Typed deserialization exactly as the DKG worker performs it: one pass, no +//! trailing bytes tolerated, followed by the param-bound structural check. +template +bool TypedDeserializeSucceeds(Span payload, const Consensus::LLMQParams& params) +{ + CDataStream ds{payload, SER_NETWORK, PROTOCOL_VERSION}; + Message msg; + try { + ds >> msg; + } catch (...) { + return false; + } + return ds.empty() && llmq::CheckDKGMessageStructure(msg, params); +} + +bool TypedDeserializeSucceeds(std::string_view msg_type, Span payload, + const Consensus::LLMQParams& params) +{ + if (msg_type == NetMsgType::QCONTRIB) { + return TypedDeserializeSucceeds(payload, params); + } else if (msg_type == NetMsgType::QCOMPLAINT) { + return TypedDeserializeSucceeds(payload, params); + } else if (msg_type == NetMsgType::QJUSTIFICATION) { + return TypedDeserializeSucceeds(payload, params); + } else if (msg_type == NetMsgType::QPCOMMITMENT) { + return TypedDeserializeSucceeds(payload, params); + } + return false; +} + +//! Build a message that is well formed for @p params, serialize it, and require +//! the framing walk to accept the bytes our own serializer just produced. This +//! covers the equivalence from the other end: it does not depend on the fuzzer +//! synthesizing a valid BLS encoding by chance, so a framing/serializer drift is +//! caught even from an empty corpus. +template +void CheckSerializedMessageIsAccepted(std::string_view msg_type, const Message& msg, + const Consensus::LLMQParams& params) +{ + CDataStream ds{SER_NETWORK, PROTOCOL_VERSION}; + ds << msg; + assert(llmq::CheckDKGMessageStructure(msg, params)); + assert(llmq::CheckDKGMessageWireStructure(msg_type, ds, params)); +} + +void CheckWellFormedMessageIsAccepted(std::string_view msg_type, const Consensus::LLMQParams& params, + FuzzedDataProvider& provider) +{ + 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; + if (size == 0 || min_size > size) { + return; + } + + if (msg_type == NetMsgType::QCONTRIB) { + llmq::CDKGContribution qc; + qc.vvec = std::make_shared>(threshold); + qc.contributions = std::make_shared>(); + const size_t blobs = provider.ConsumeIntegralInRange(min_size, size); + qc.contributions->blobs.resize(blobs); + for (auto& blob : qc.contributions->blobs) { + blob.resize(provider.ConsumeIntegralInRange(0, 64)); + } + CheckSerializedMessageIsAccepted(msg_type, qc, params); + } else if (msg_type == NetMsgType::QCOMPLAINT) { + llmq::CDKGComplaint qc; + const size_t members = provider.ConsumeIntegralInRange(0, size); + qc.badMembers.assign(members, false); + qc.complainForMembers.assign(members, false); + for (size_t i = 0; i < members; ++i) { + qc.badMembers[i] = provider.ConsumeBool(); + qc.complainForMembers[i] = provider.ConsumeBool(); + } + CheckSerializedMessageIsAccepted(msg_type, qc, params); + } else if (msg_type == NetMsgType::QJUSTIFICATION) { + llmq::CDKGJustification qj; + qj.contributions.resize(provider.ConsumeIntegralInRange(0, size)); + CheckSerializedMessageIsAccepted(msg_type, qj, params); + } else if (msg_type == NetMsgType::QPCOMMITMENT) { + llmq::CDKGPrematureCommitment qc; + const size_t members = provider.ConsumeIntegralInRange(0, size); + qc.validMembers.assign(members, false); + for (size_t i = 0; i < members; ++i) { + qc.validMembers[i] = provider.ConsumeBool(); + } + CheckSerializedMessageIsAccepted(msg_type, qc, params); + } +} + +void initialize_dkg_message_framing() +{ + BLSInit(); +} +} // namespace + +/** + * DKG network intake validates framing without decoding BLS objects, then the + * DKG worker deserializes the retained bytes exactly once. Those are two + * hand-maintained parsers over one wire format, so they can drift apart. + * + * The safety-critical direction is that the framing walk must never reject a + * payload the worker would have accepted -- otherwise honest DKG messages are + * silently dropped at intake and quorum formation degrades. The converse is not + * asserted: framing deliberately accepts payloads whose BLS points fail to + * decode, so that the worker can score the sender. + */ +FUZZ_TARGET(dkg_message_framing, .init = initialize_dkg_message_framing) +{ + FuzzedDataProvider fuzzed_data_provider{buffer.data(), buffer.size()}; + + const std::string_view msg_type = DKG_MSG_TYPES.at( + fuzzed_data_provider.ConsumeIntegralInRange(0, DKG_MSG_TYPES.size() - 1)); + const Consensus::LLMQParams& params = Consensus::available_llmqs.at( + fuzzed_data_provider.ConsumeIntegralInRange(0, Consensus::available_llmqs.size() - 1)); + // Framing is scheme-independent (BLS wire sizes are fixed), but typed + // deserialization is not; cover both so the equivalence is checked on each. + bls::bls_legacy_scheme.store(fuzzed_data_provider.ConsumeBool()); + + CheckWellFormedMessageIsAccepted(msg_type, params, fuzzed_data_provider); + + const std::vector payload = fuzzed_data_provider.ConsumeRemainingBytes(); + + const CDataStream payload_stream{payload, SER_NETWORK, PROTOCOL_VERSION}; + const bool framing_ok = llmq::CheckDKGMessageWireStructure(msg_type, payload_stream, params); + const bool typed_ok = TypedDeserializeSucceeds(msg_type, payload, params); + + assert(!typed_ok || framing_ok); +} diff --git a/test/util/data/non-backported.txt b/test/util/data/non-backported.txt index f759bb01a1e1..f7482bfb0da7 100644 --- a/test/util/data/non-backported.txt +++ b/test/util/data/non-backported.txt @@ -65,6 +65,7 @@ src/test/dip0020opcodes_tests.cpp src/test/dip14_tests.cpp src/test/dynamic_activation*.cpp src/test/evo*.cpp +src/test/fuzz/dkg_message_framing.cpp src/test/llmq*.cpp src/test/masternode_payments_tests.cpp src/test/spork_tests.cpp From 5d4656170d9080c83a8a273b24bbd96987fd41c7 Mon Sep 17 00:00:00 2001 From: pasta Date: Fri, 7 Aug 2026 15:43:27 -0500 Subject: [PATCH 3/3] test: cover DKG pending-queue quotas and worker-deferred BLS rejection Unit tests pin the CDKGPendingMessages semantics: the per-proTx quota survives reconnects and is not refunded by drains, duplicates are rejected before charging, quotas are independent across proTxes, and own messages are charged under this node's own proTxHash. Functional tests cover trailing-byte rejection at intake, deferral of BLS decoding to the DKG worker (scored there, not at intake), quota persistence across reconnects under fresh NodeIds, and late-message retention cleared at round start without BLS work. Co-Authored-By: Claude Fable 5 --- test/functional/feature_llmq_dkg_intake.py | 123 +++++++++++++++++---- 1 file changed, 101 insertions(+), 22 deletions(-) diff --git a/test/functional/feature_llmq_dkg_intake.py b/test/functional/feature_llmq_dkg_intake.py index 514ef954cd43..e6c96e9d9f4f 100755 --- a/test/functional/feature_llmq_dkg_intake.py +++ b/test/functional/feature_llmq_dkg_intake.py @@ -6,18 +6,25 @@ feature_llmq_dkg_intake.py Adversarial P2P tests for DKG message-intake hardening: + - observer nodes neither request nor retain DKG round payloads because they + have no DKG worker to consume them. - pushed DKG messages (qcontrib/qcomplaint/qjustify/qpcommit) from a peer that is 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. + - framing pre-validation: truncated, trailing, or parametrically out-of-bounds + DKG payloads are rejected before retention even from a verified peer. + - BLS objects are not decoded at intake: a framing-valid payload with an invalid + BLS encoding reaches the DKG worker and is rejected (and scored) there. - the per-proTx retention quota is keyed by the MNAuth-verified proTxHash, so reconnecting under a fresh NodeId does not refill it. + - late, framing-valid messages are discarded without BLS decoding before the + next round initializes. - a well-formed DKG message that the peer never announced and was never asked for is dropped before retention, even from a verified peer. -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 ( @@ -42,6 +49,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 # LLMQ_TEST dkgInterval; phaseBlocks=2, so stage 0=Initialized, 2=Contribute, 4=Complain. CYCLE_LENGTH = 24 @@ -111,8 +120,14 @@ 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, while - # -debug=llmq-dkg exposes queue-boundary behavior (quota drops). - extra_args = [["-whitelist=127.0.0.1", "-debug=net", "-debug=llmq-dkg", "-deprecatedrpc=banscore"]] * 4 + # -debug=llmq-dkg exposes worker and queue-boundary behavior. + common_args = [ + "-whitelist=127.0.0.1", + "-debug=net", + "-debug=llmq-dkg", + "-deprecatedrpc=banscore", + ] + extra_args = [common_args + ["-watchquorums=1"]] + [common_args] * 3 self.set_dash_test_params(4, 3, extra_args=extra_args) def quorum_hash_prefix(self): @@ -121,14 +136,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, protx_hash=0): + 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(protx_hash) - 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): @@ -153,13 +168,37 @@ def run_test(self): # Target an active masternode -- the realistic victim of these messages. mn_node = self.mninfo[0].get_node(self) + self.test_observer_drops_dkg_messages(node0) 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(mn_node) self.test_under_min_contribution_blobs_rejected(mn_node) self.test_unrequested_rejected(mn_node) + def test_observer_drops_dkg_messages(self, node): + self.log.info("An observer neither requests nor retains DKG round messages") + peer, peer_id = self.add_verified_peer(node) + wait_for_banscore(node, peer_id, 0) + + payload = self.qcontrib_payload( + blob_count=2, + vvec_pubkey=INVALID_NONZERO_BLS_PUBKEY, + ) + inv_hash = uint256_from_str(hash256(payload)) + peer.last_message.pop("getdata", None) + peer.send_message(msg_inv([CInv(MSG_QUORUM_CONTRIB, inv_hash)])) + peer.sync_with_ping() + assert "getdata" not in peer.last_message + + with node.assert_debug_log(["ignoring qcontrib in observer mode"]): + peer.send_message(msg_dkg_raw(b"qcontrib", payload)) + peer.sync_with_ping() + wait_for_banscore(node, peer_id, 0) + node.disconnect_p2ps() + def test_unverified_sender_rejected(self, node): self.log.info("Pushed DKG messages from a non-verified peer are rejected (Misbehaving 10 each)") peer = node.add_p2p_connection(P2PInterface()) @@ -191,8 +230,8 @@ def test_malformed_rejected(self, node): self.log.info("Malformed DKG payloads are rejected even from a verified peer (Misbehaving 100)") peer, peer_id = self.add_verified_peer(node) wait_for_banscore(node, peer_id, 0) - # Valid llmqType + quorumHash prefix, then too few bytes to deserialize a - # CDKGContribution -> structural pre-validation rejects it before retention. + # Valid llmqType + quorumHash prefix, then too few bytes to walk a + # CDKGContribution -> framing pre-validation rejects it before retention. payload = self.quorum_hash_prefix() + b"\x00\x00\x00\x00" with node.assert_debug_log(["malformed DKG message"]): peer.send_message(msg_dkg_raw(b"qcontrib", payload)) @@ -200,6 +239,19 @@ 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).""" skip_count = CYCLE_LENGTH - (self.nodes[0].getblockcount() % CYCLE_LENGTH) @@ -208,16 +260,40 @@ def _start_fresh_dkg_cycle(self, 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 _send_late_qcontrib(self, peer, nonce): - """Send a well-formed QCONTRIB that no on-time worker will drain. + 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) - Unique proTxHash bytes per message so retention is bounded by the quotas - rather than by duplicate-hash suppression. - """ - send_requested_qcontrib(peer, self.qcontrib_payload(blob_count=2, protx_hash=nonce)) + 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", "unrequested DKG message"], + timeout=10, + ): + send_requested_qcontrib(peer, self.qcontrib_payload( + blob_count=2, + vvec_pubkey=INVALID_NONZERO_BLS_PUBKEY, + )) + wait_for_banscore(node, peer_id, 0) + self.move_blocks(nodes, 2) + wait_for_banscore(node, peer_id, 100) + node.disconnect_p2ps() + + def _send_late_qcontrib(self, peer, nonce): + """Send a framing-valid, BLS-invalid QCONTRIB that no on-time worker will drain.""" + send_requested_qcontrib(peer, self.qcontrib_payload( + blob_count=2, + vvec_pubkey=INVALID_NONZERO_BLS_PUBKEY, + # Unique bytes per message so retention is bounded by the quotas rather + # than by duplicate-hash suppression. + protx_hash=nonce, + )) def test_late_messages_bounded(self, node): - self.log.info("Late QCONTRIB retention is bounded per proTx, then cleared at round start") + self.log.info("Late QCONTRIB retention is bounded per proTx, then cleared without BLS decoding") nodes = [self.nodes[0]] + [mn.get_node(self) for mn in self.mninfo] self._start_fresh_dkg_cycle(nodes) stage = self.nodes[0].getblockcount() % CYCLE_LENGTH @@ -246,19 +322,22 @@ def test_late_messages_bounded(self, node): wait_for_banscore(node, quota_peer_id, 0) # A distinct proTx has its own quota. Keep it connected to verify that - # round-start clearing does not score its retained message. + # round-start clearing does not score its stale BLS encoding. nonce += 1 retained_peer, retained_peer_id = self.add_verified_peer(node, "dkg-retained", protx="%064x" % 0xd0) self._send_late_qcontrib(retained_peer, nonce) wait_for_banscore(node, retained_peer_id, 0) - # Crossing the round boundary must clear the raw queue; the retained - # messages must never reach a worker (whose preverification would score - # their unknown proTxHashes). + # Crossing the phase boundary must clear the 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=["failed preverification"], + unexpected_msgs=[ + "malformed DKG message", + "failed to deserialize message", + "message failed structure check", + ], timeout=60, ): self.move_blocks(nodes, remaining)