fix(net): bound CoinJoin message vector intake

src/coinjoin/coinjoin.cpp

@@ -95,7 +95,7 @@ bool CCoinJoinBroadcastTx::IsValidStructure() const
     if (tx->vin.size() < size_t(CoinJoin::GetMinPoolParticipants())) {
         return false;
     }
-    if (tx->vin.size() > CoinJoin::GetMaxPoolParticipants() * COINJOIN_ENTRY_MAX_SIZE) {
+    if (tx->vin.size() > CoinJoin::GetMaxPoolInputOutputCount()) {
         return false;
     }
     return std::ranges::all_of(tx->vout, [](const auto& txOut) {

src/coinjoin/coinjoin.h

@@ -13,6 +13,7 @@
 #include <netaddress.h>
 #include <primitives/block.h>
 #include <primitives/transaction.h>
+#include <serialize.h>
 #include <sync.h>
 #include <timedata.h>
 #include <util/translation.h>
@@ -49,6 +50,15 @@ static constexpr int COINJOIN_SIGNING_TIMEOUT = 15;
 
 static constexpr size_t COINJOIN_ENTRY_MAX_SIZE = 9;
 
+namespace CoinJoin {
+/// Get the minimum/maximum number of participants for the pool
+int GetMinPoolParticipants();
+int GetMaxPoolParticipants();
+
+/// Maximum number of inputs or outputs across a full pool
+inline size_t GetMaxPoolInputOutputCount() { return size_t(GetMaxPoolParticipants()) * COINJOIN_ENTRY_MAX_SIZE; }
+} // namespace CoinJoin
+
 // pool responses
 enum PoolMessage : int32_t {
     ERR_ALREADY_HAVE,
@@ -167,9 +177,25 @@ class CCoinJoinEntry
     {
     }
 
-    SERIALIZE_METHODS(CCoinJoinEntry, obj)
+    template <typename Stream>
+    void Serialize(Stream& s) const
+    {
+        s << vecTxDSIn << txCollateral << vecTxOut;
+    }
+
+    template <typename Stream>
+    void Unserialize(Stream& s)
     {
-        READWRITE(obj.vecTxDSIn, obj.txCollateral, obj.vecTxOut);
+        const size_t max_count{CoinJoin::GetMaxPoolInputOutputCount()};
+        if (!UnserializeVectorWithMaxSize(s, vecTxDSIn, max_count)) {
+            throw std::ios_base::failure("CCoinJoinEntry::vecTxDSIn size too large");
+        }
+
+        s >> txCollateral;
+
+        if (!UnserializeVectorWithMaxSize(s, vecTxOut, max_count)) {
+            throw std::ios_base::failure("CCoinJoinEntry::vecTxOut size too large");
+        }
     }
 
     bool AddScriptSig(const CTxIn& txin);
@@ -380,10 +406,6 @@ namespace CoinJoin
 {
     bilingual_str GetMessageByID(PoolMessage nMessageID);
 
-    /// Get the minimum/maximum number of participants for the pool
-    int GetMinPoolParticipants();
-    int GetMaxPoolParticipants();
-
     constexpr CAmount GetMaxPoolAmount() { return COINJOIN_ENTRY_MAX_SIZE * vecStandardDenominations.front(); }
 
     /// If the collateral is valid given by a client

src/coinjoin/server.cpp

@@ -15,6 +15,7 @@
 #include <netmessagemaker.h>
 #include <scheduler.h>
 #include <script/interpreter.h>
+#include <serialize.h>
 #include <shutdown.h>
 #include <streams.h>
 #include <txmempool.h>
@@ -58,7 +59,7 @@ void CCoinJoinServer::ProcessMessage(CNode& peer, const std::string& msg_type, C
     } else if (msg_type == NetMsgType::DSVIN) {
         ProcessDSVIN(peer, vRecv);
     } else if (msg_type == NetMsgType::DSSIGNFINALTX) {
-        ProcessDSSIGNFINALTX(vRecv);
+        ProcessDSSIGNFINALTX(peer, vRecv);
     }
 }
 
@@ -222,10 +223,40 @@ void CCoinJoinServer::ProcessDSVIN(CNode& peer, CDataStream& vRecv)
     }
 }
 
-void CCoinJoinServer::ProcessDSSIGNFINALTX(CDataStream& vRecv)
+void CCoinJoinServer::ProcessDSSIGNFINALTX(CNode& peer, CDataStream& vRecv)
 {
+    // Only accept signatures while we are actually collecting them, and only
+    // from peers that are active participants in this session. Otherwise a
+    // stray or unauthenticated peer could abort the session for everyone.
+    if (nState != POOL_STATE_SIGNING) {
+        LogPrint(BCLog::COINJOIN, "DSSIGNFINALTX -- wrong state, nState=%d, peer=%d\n",
+                 nState.load(), peer.GetId());
+        PushStatus(peer, STATUS_REJECTED, ERR_SESSION);
+        return;
+    }
+    {
+        LOCK(cs_coinjoin);
+        const bool is_participant = std::ranges::any_of(
+            vecEntries, [&peer](const auto& entry) { return entry.addr == peer.addr; });
+        if (!is_participant) {
+            LogPrint(BCLog::COINJOIN, "DSSIGNFINALTX -- ignoring message from non-participant peer=%d\n",
+                     peer.GetId());
+            PushStatus(peer, STATUS_REJECTED, ERR_INVALID_INPUT);
+            return;
+        }
+    }
+
+    const size_t max_txins{CoinJoin::GetMaxPoolInputOutputCount()};
     std::vector<CTxIn> vecTxIn;
-    vRecv >> vecTxIn;
+    // Reject an over-cap count through this peer-local ERR_MAXIMUM path before a
+    // single CTxIn is decoded or allocated. A count above the generic MAX_SIZE cap
+    // is malformed and throws from ReadCompactSize, as it does for any other message.
+    if (!UnserializeVectorWithMaxSize(vRecv, vecTxIn, max_txins)) {
+        LogPrint(BCLog::COINJOIN, "DSSIGNFINALTX -- too many inputs. max=%d, peer=%d\n",
+                 static_cast<int>(max_txins), peer.GetId());
+        PushStatus(peer, STATUS_REJECTED, ERR_MAXIMUM);
+        return;
+    }
 
     LogPrint(BCLog::COINJOIN, "DSSIGNFINALTX -- vecTxIn.size() %s\n", vecTxIn.size());
 
@@ -579,16 +610,31 @@ bool CCoinJoinServer::AddEntry(const CCoinJoinEntry& entry, PoolMessage& nMessag
         return false;
     }
 
-    if (!CoinJoin::IsCollateralValid(m_chainman, m_isman, mempool, *entry.txCollateral)) {
-        LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: collateral not valid!\n", __func__);
-        nMessageIDRet = ERR_INVALID_COLLATERAL;
+    if (entry.vecTxDSIn.size() > COINJOIN_ENTRY_MAX_SIZE || entry.vecTxOut.size() > COINJOIN_ENTRY_MAX_SIZE) {
+        LogPrint(BCLog::COINJOIN, /* Continued */
+                 "CCoinJoinServer::%s -- ERROR: too many inputs or outputs! inputs=%s/%s, outputs=%s/%s\n", __func__,
+                 entry.vecTxDSIn.size(), COINJOIN_ENTRY_MAX_SIZE, entry.vecTxOut.size(), COINJOIN_ENTRY_MAX_SIZE);
+        nMessageIDRet = ERR_MAXIMUM;
+
+        CTransactionRef txCollateralToConsume;
+        {
+            LOCK(cs_coinjoin);
+            const auto it = std::ranges::find_if(vecSessionCollaterals, [&entry](const auto& txCollateral) {
+                return *entry.txCollateral == *txCollateral;
+            });
+            if (it != vecSessionCollaterals.end()) {
+                txCollateralToConsume = *it;
+            }
+        }
+        if (txCollateralToConsume) {
+            ConsumeCollateral(txCollateralToConsume);
+        }
         return false;
     }
 
-    if (entry.vecTxDSIn.size() > COINJOIN_ENTRY_MAX_SIZE) {
-        LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: too many inputs! %d/%d\n", __func__, entry.vecTxDSIn.size(), COINJOIN_ENTRY_MAX_SIZE);
-        nMessageIDRet = ERR_MAXIMUM;
-        ConsumeCollateral(entry.txCollateral);
+    if (!CoinJoin::IsCollateralValid(m_chainman, m_isman, mempool, *entry.txCollateral)) {
+        LogPrint(BCLog::COINJOIN, "CCoinJoinServer::%s -- ERROR: collateral not valid!\n", __func__);
+        nMessageIDRet = ERR_INVALID_COLLATERAL;
         return false;
     }
 

src/coinjoin/server.h

@@ -88,7 +88,7 @@ class CCoinJoinServer : public CCoinJoinBaseSession, public NetHandler
     void ProcessDSACCEPT(CNode& peer, CDataStream& vRecv);
     void ProcessDSQUEUE(NodeId from, CDataStream& vRecv);
     void ProcessDSVIN(CNode& peer, CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin);
-    void ProcessDSSIGNFINALTX(CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin);
+    void ProcessDSSIGNFINALTX(CNode& peer, CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin);
 
     void SetNull() override EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin);
 

src/test/coinjoin_inouts_tests.cpp

@@ -2,26 +2,42 @@
 // Distributed under the MIT/X11 software license, see the accompanying
 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
-#include <test/util/setup_common.h>
-
-#include <algorithm>
-#include <array>
-#include <cstdint>
-#include <vector>
-
+#include <active/masternode.h>
+#include <bls/bls.h>
 #include <chain.h>
 #include <coinjoin/coinjoin.h>
 #include <coinjoin/common.h>
+#include <coinjoin/server.h>
 #include <evo/chainhelper.h>
+#include <llmq/context.h>
+#include <masternode/sync.h>
+#include <net.h>
+#include <node/connection_types.h>
+#include <protocol.h>
 #include <script/script.h>
+#include <streams.h>
+#include <test/util/setup_common.h>
 #include <uint256.h>
 #include <util/check.h>
 #include <util/time.h>
 
 #include <boost/test/unit_test.hpp>
 
+#include <algorithm>
+#include <array>
+#include <cstdint>
+#include <memory>
+#include <vector>
+
 BOOST_FIXTURE_TEST_SUITE(coinjoin_inouts_tests, TestingSetup)
 
+static CBLSSecretKey MakeSecretKey()
+{
+    CBLSSecretKey sk;
+    sk.MakeNewKey();
+    return sk;
+}
+
 static CScript P2PKHScript(uint8_t tag = 0x01)
 {
     // OP_DUP OP_HASH160 <20-byte-tag> OP_EQUALVERIFY OP_CHECKSIG
@@ -155,6 +171,169 @@ BOOST_AUTO_TEST_CASE(entry_addscriptsig_matches_and_rejects)
     }
 }
 
+// Test-only subclass exposing the minimal seams needed to observe how
+// ProcessDSSIGNFINALTX treats messages from participants vs. non-participants
+// without standing up a full DKG-backed signing session.
+class TestableCoinJoinServer : public CCoinJoinServer
+{
+public:
+    using CCoinJoinServer::CCoinJoinServer;
+
+    void EnterSigningState() { nState = POOL_STATE_SIGNING; }
+
+    void SeedParticipant(const CService& addr) EXCLUSIVE_LOCKS_REQUIRED(!cs_coinjoin)
+    {
+        CCoinJoinEntry entry;
+        entry.addr = addr;
+        LOCK(cs_coinjoin);
+        vecEntries.push_back(std::move(entry));
+    }
+};
+
+static std::unique_ptr<CNode> MakePeer(NodeId id, uint32_t ipv4)
+{
+    in_addr peer_in_addr{};
+    peer_in_addr.s_addr = htonl(ipv4);
+    auto peer = std::make_unique<CNode>(id,
+                                        /*sock=*/nullptr,
+                                        /*addrIn=*/CAddress{CService{peer_in_addr, 8333}, NODE_NETWORK},
+                                        /*nKeyedNetGroupIn=*/0,
+                                        /*nLocalHostNonceIn=*/0,
+                                        /*addrBindIn=*/CAddress{},
+                                        /*addrNameIn=*/std::string{},
+                                        /*conn_type_in=*/ConnectionType::INBOUND,
+                                        /*inbound_onion=*/false);
+    peer->nVersion = PROTOCOL_VERSION;
+    peer->SetCommonVersion(PROTOCOL_VERSION);
+    return peer;
+}
+
+BOOST_AUTO_TEST_CASE(server_signfinaltx_nonparticipant_cannot_abort_session)
+{
+    BOOST_REQUIRE(m_node.mn_sync);
+    m_node.mn_sync->SwitchToNextAsset();
+    BOOST_REQUIRE(m_node.mn_sync->IsBlockchainSynced());
+
+    CActiveMasternodeManager mn_activeman(*Assert(m_node.connman), *Assert(m_node.dmnman), MakeSecretKey());
+    TestableCoinJoinServer server(m_node.peerman.get(), *Assert(m_node.chainman), *Assert(m_node.connman),
+                                  *Assert(m_node.dmnman), *Assert(m_node.dstxman), *Assert(m_node.mn_metaman),
+                                  *Assert(m_node.mempool), mn_activeman, *Assert(m_node.mn_sync),
+                                  *Assert(m_node.llmq_ctx->isman));
+
+    // Seed an active signing session with one participant. That participant's
+    // addr is deliberately not registered with connman -- a session-wide
+    // RelayStatus(REJECTED) would therefore see nDisconnected == vecEntries.size()
+    // and reset the pool state to POOL_STATE_IDLE via SetNull().
+    auto participant = MakePeer(/*id=*/7, /*ipv4=*/0x0a000001);
+    server.SeedParticipant(participant->addr);
+    server.EnterSigningState();
+    BOOST_REQUIRE_EQUAL(server.GetState(), int{POOL_STATE_SIGNING});
+
+    auto nonparticipant = MakePeer(/*id=*/42, /*ipv4=*/0x01020304);
+    BOOST_REQUIRE(!(nonparticipant->addr == participant->addr));
+
+    const size_t max_txins{CoinJoin::GetMaxPoolInputOutputCount()};
+    CDataStream stream{SER_NETWORK, PROTOCOL_VERSION};
+    WriteCompactSize(stream, max_txins + 1);
+
+    BOOST_CHECK_NO_THROW(server.ProcessMessage(*nonparticipant, NetMsgType::DSSIGNFINALTX, stream));
+
+    // The non-participant must be rejected without touching session-wide state:
+    // the stream body must not have been read past the compact-size prefix, and
+    // the pool must still be in POOL_STATE_SIGNING with its entry intact.
+    BOOST_CHECK_EQUAL(stream.size(), GetSizeOfCompactSize(max_txins + 1));
+    BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_SIGNING});
+    BOOST_CHECK_EQUAL(server.GetEntriesCount(), 1);
+}
+
+BOOST_AUTO_TEST_CASE(server_signfinaltx_participant_oversized_count_is_rejected_locally)
+{
+    BOOST_REQUIRE(m_node.mn_sync);
+    m_node.mn_sync->SwitchToNextAsset();
+    BOOST_REQUIRE(m_node.mn_sync->IsBlockchainSynced());
+
+    CActiveMasternodeManager mn_activeman(*Assert(m_node.connman), *Assert(m_node.dmnman), MakeSecretKey());
+    TestableCoinJoinServer server(m_node.peerman.get(), *Assert(m_node.chainman), *Assert(m_node.connman),
+                                  *Assert(m_node.dmnman), *Assert(m_node.dstxman), *Assert(m_node.mn_metaman),
+                                  *Assert(m_node.mempool), mn_activeman, *Assert(m_node.mn_sync),
+                                  *Assert(m_node.llmq_ctx->isman));
+
+    // Same setup, but this time the oversized DSSIGNFINALTX comes from the
+    // session participant itself. It must still be rejected without materializing
+    // the txin vector and without collapsing the session for everyone else.
+    auto participant = MakePeer(/*id=*/7, /*ipv4=*/0x0a000001);
+    server.SeedParticipant(participant->addr);
+    server.EnterSigningState();
+
+    const size_t max_txins{CoinJoin::GetMaxPoolInputOutputCount()};
+    CDataStream stream{SER_NETWORK, PROTOCOL_VERSION};
+    WriteCompactSize(stream, max_txins + 1);
+
+    BOOST_CHECK_NO_THROW(server.ProcessMessage(*participant, NetMsgType::DSSIGNFINALTX, stream));
+    BOOST_CHECK_EQUAL(stream.size(), 0U);
+    BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_SIGNING});
+    BOOST_CHECK_EQUAL(server.GetEntriesCount(), 1);
+
+    // A count beyond the generic CompactSize cap is a malformed message, not a
+    // CoinJoin-level violation: it throws the standard deserialization error out
+    // to net processing. No txin is materialized and the session is left intact,
+    // so honest participants are unaffected.
+    CDataStream huge_stream{SER_NETWORK, PROTOCOL_VERSION};
+    WriteCompactSize(huge_stream, uint64_t{MAX_SIZE} + 1);
+
+    BOOST_CHECK_THROW(server.ProcessMessage(*participant, NetMsgType::DSSIGNFINALTX, huge_stream),
+                      std::ios_base::failure);
+    BOOST_CHECK_EQUAL(server.GetState(), int{POOL_STATE_SIGNING});
+    BOOST_CHECK_EQUAL(server.GetEntriesCount(), 1);
+}
+
+BOOST_AUTO_TEST_CASE(entry_deserializes_vectors_through_wire_cap)
+{
+    const size_t wire_cap{CoinJoin::GetMaxPoolInputOutputCount()};
+    BOOST_REQUIRE_GT(wire_cap, COINJOIN_ENTRY_MAX_SIZE);
+
+    for (const size_t count : {size_t{0}, COINJOIN_ENTRY_MAX_SIZE, COINJOIN_ENTRY_MAX_SIZE + 1, wire_cap}) {
+        BOOST_TEST_CONTEXT("count=" << count)
+        {
+            CCoinJoinEntry entry;
+            entry.vecTxDSIn.resize(count);
+            entry.vecTxOut.resize(count);
+
+            CDataStream stream{SER_NETWORK, PROTOCOL_VERSION};
+            stream << entry;
+
+            CCoinJoinEntry roundtripped;
+            BOOST_CHECK_NO_THROW(stream >> roundtripped);
+            BOOST_CHECK_EQUAL(roundtripped.vecTxDSIn.size(), count);
+            BOOST_CHECK_EQUAL(roundtripped.vecTxOut.size(), count);
+        }
+    }
+}
+
+BOOST_AUTO_TEST_CASE(entry_rejects_inputs_above_wire_cap_before_materializing)
+{
+    const size_t wire_cap{CoinJoin::GetMaxPoolInputOutputCount()};
+    CDataStream stream{SER_NETWORK, PROTOCOL_VERSION};
+    WriteCompactSize(stream, wire_cap + 1);
+
+    CCoinJoinEntry entry;
+    BOOST_CHECK_THROW(stream >> entry, std::ios_base::failure);
+    BOOST_CHECK(entry.vecTxDSIn.empty());
+}
+
+BOOST_AUTO_TEST_CASE(entry_rejects_outputs_above_wire_cap_before_materializing)
+{
+    const size_t wire_cap{CoinJoin::GetMaxPoolInputOutputCount()};
+    CDataStream stream{SER_NETWORK, PROTOCOL_VERSION};
+    WriteCompactSize(stream, 0);
+    stream << MakeTransactionRef(CMutableTransaction{});
+    WriteCompactSize(stream, wire_cap + 1);
+
+    CCoinJoinEntry entry;
+    BOOST_CHECK_THROW(stream >> entry, std::ios_base::failure);
+    BOOST_CHECK(entry.vecTxOut.empty());
+}
+
 BOOST_AUTO_TEST_CASE(queue_timeout_bounds)
 {
     CCoinJoinQueue dsq;