Merge #7347: fix: punish invalid dstx messages

4329a4f refactor: use named result type and score enum for ValidateDSTX (PastaClaw)
7d073b9 fix: penalize unknown-masternode dstx relays with a small score (PastaClaw)
9c68646 fix: avoid punishing unverifiable dstx relays (PastaClaw)
c25cf3d fix: punish invalid dstx messages (PastaClaw)

Pull request description:

  # fix: punish invalid dstx messages

  ## Issue being fixed or feature implemented

  Invalid CoinJoin broadcast transaction (`dstx`) messages can return early from
  validation without applying the same peer-accounting consequence used by other
  invalid P2P messages.

  This hardens DSTX handling by making invalid validation results contribute a low
  misbehavior score while preserving benign early returns for duplicate or
  otherwise non-error DSTX cases.

  ## What was done?

  - Apply a low misbehavior score when DSTX validation returns an invalid/error
    result.
  - Preserve existing behavior for successful validation and benign early-return
    cases.
  - Add P2P test framework serialization for `dstx` messages.
  - Add a focused functional test covering invalid DSTX peer accounting.

  ## How Has This Been Tested?

  Tested on macOS arm64.

  - `git diff --check`
  - `python3 -m py_compile test/functional/p2p_dstx.py test/functional/test_framework/messages.py`
  - Configured with depends `config.site`, `--without-gui`, `--disable-bench`,
    and `--disable-fuzz-binary`
  - `make -j8 src/dashd`
  - `test/functional/p2p_dstx.py --tmpdir=/tmp/dash_func_dstx`
  - `test/functional/p2p_dstx.py --tmpdir=/tmp/dash_func_dstx_2`

  ## Breaking Changes

  None.

  ## Checklist

  - [x] I have performed a self-review of my own code
  - [x] I have commented my code, particularly in hard-to-understand areas
  - [x] I have added or updated relevant unit/integration/functional/e2e tests
  - [ ] I have made corresponding changes to the documentation
  - [ ] I have assigned this pull request to a milestone

ACKs for top commit:
  UdjinM6:
    utACK 4329a4f

Tree-SHA512: 79ac5b9bf9d359d68abf9b8a890fd09719c642635f199568efe4cb61c9979aa8cf89df48e6229c5c807b12b4ae127be701aef95ec04f4518fcabcdde305ce9b0

Author: PastaPastaPasta

src/net_processing.cpp

@@ -3490,18 +3490,31 @@ void PeerManagerImpl::ProcessGetCFCheckPt(CNode& node, Peer& peer, CDataStream&
     m_connman.PushMessage(&node, std::move(msg));
 }
 
-std::pair<bool /*ret*/, bool /*do_return*/> static ValidateDSTX(CDeterministicMNManager& dmnman, CDSTXManager& dstxman, ChainstateManager& chainman,
-                                                                CMasternodeMetaMan& mn_metaman, CTxMemPool& mempool, CCoinJoinBroadcastTx& dstx, uint256 hashTx)
+// Misbehavior penalty to apply to the relaying peer; NONE means no penalty.
+enum class DSTXValidationScore : int {
+    NONE = 0,
+    UNKNOWN_MASTERNODE = 1,
+    INVALID = 10,
+};
+
+// do_return signals the caller to stop further processing of the DSTX.
+struct DSTXValidationResult {
+    DSTXValidationScore score;
+    bool do_return;
+};
+
+static DSTXValidationResult ValidateDSTX(CDeterministicMNManager& dmnman, CDSTXManager& dstxman, ChainstateManager& chainman,
+                                         CMasternodeMetaMan& mn_metaman, CTxMemPool& mempool, CCoinJoinBroadcastTx& dstx, uint256 hashTx)
 {
     assert(mn_metaman.IsValid());
 
     if (!dstx.IsValidStructure()) {
         LogPrint(BCLog::COINJOIN, "DSTX -- Invalid DSTX structure: %s\n", hashTx.ToString());
-        return {false, true};
+        return {DSTXValidationScore::INVALID, true};
     }
     if (dstxman.GetDSTX(hashTx)) {
         LogPrint(BCLog::COINJOIN, "DSTX -- Already have %s, skipping...\n", hashTx.ToString());
-        return {true, true}; // not an error
+        return {DSTXValidationScore::NONE, true}; // not an error
     }
 
     const CBlockIndex* pindex{nullptr};
@@ -3534,26 +3547,29 @@ std::pair<bool /*ret*/, bool /*do_return*/> static ValidateDSTX(CDeterministicMN
 
     if (!dmn) {
         LogPrint(BCLog::COINJOIN, "DSTX -- Can't find masternode %s to verify %s\n", dstx.masternodeOutpoint.ToStringShort(), hashTx.ToString());
-        return {false, true};
+        // We can't verify the signature here, so apply only a small penalty.
+        // The MN may have been removed very recently, but a peer flooding us with
+        // unverifiable DSTX-es should still eventually be discouraged.
+        return {DSTXValidationScore::UNKNOWN_MASTERNODE, true};
     }
 
     if (!mn_metaman.IsValidForMixingTxes(dmn->proTxHash)) {
         LogPrint(BCLog::COINJOIN, "DSTX -- Masternode %s is sending too many transactions %s\n", dstx.masternodeOutpoint.ToStringShort(), hashTx.ToString());
-        return {true, true};
+        return {DSTXValidationScore::NONE, true};
         // TODO: Not an error? Could it be that someone is relaying old DSTXes
         // we have no idea about (e.g we were offline)? How to handle them?
     }
 
     if (!dstx.CheckSignature(dmn->pdmnState->pubKeyOperator.Get())) {
         LogPrint(BCLog::COINJOIN, "DSTX -- CheckSignature() failed for %s\n", hashTx.ToString());
-        return {false, true};
+        return {DSTXValidationScore::INVALID, true};
     }
 
     LogPrint(BCLog::COINJOIN, "DSTX -- Got Masternode transaction %s\n", hashTx.ToString());
     mempool.PrioritiseTransaction(hashTx, 0.1*COIN);
     mn_metaman.DisallowMixing(dmn->proTxHash);
 
-    return {true, false};
+    return {DSTXValidationScore::NONE, false};
 }
 
 void PeerManagerImpl::ProcessBlock(CNode& node, const std::shared_ptr<const CBlock>& block, bool force_processing)
@@ -4617,12 +4633,14 @@ void PeerManagerImpl::ProcessMessage(
 
         // Process custom logic, no matter if tx will be accepted to mempool later or not
         if (nInvType == MSG_DSTX) {
-           // Validate DSTX and return bRet if we need to return from here
-           uint256 hashTx = tx.GetHash();
-           const auto& [bRet, bDoReturn] = ValidateDSTX(*m_dmnman, m_dstxman, m_chainman, m_mn_metaman, m_mempool, dstx, hashTx);
-           if (bDoReturn) {
-               return;
-           }
+            uint256 hashTx = tx.GetHash();
+            const auto result = ValidateDSTX(*m_dmnman, m_dstxman, m_chainman, m_mn_metaman, m_mempool, dstx, hashTx);
+            if (result.do_return) {
+                if (result.score != DSTXValidationScore::NONE) {
+                    Misbehaving(pfrom.GetId(), static_cast<int>(result.score), "invalid dstx");
+                }
+                return;
+            }
         }
 
         LOCK(cs_main);

test/functional/p2p_dstx.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+# Copyright (c) 2026 The Dash Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Test P2P CoinJoin broadcast transaction handling.
+
+Verifies that DSTX messages with an unverifiable (unknown) masternode incur
+only a small misbehavior penalty, while clearly malformed DSTXes get the
+existing stronger penalty. Also exercises the cumulative behavior so that
+a peer flooding unknown-MN DSTXes is eventually discouraged.
+"""
+
+import time
+
+from test_framework.messages import (
+    CCoinJoinBroadcastTx,
+    COIN,
+    COutPoint,
+    CTransaction,
+    CTxIn,
+    CTxOut,
+    msg_dstx,
+)
+from test_framework.p2p import P2PInterface
+from test_framework.script import (
+    CScript,
+    OP_CHECKSIG,
+    OP_DUP,
+    OP_EQUALVERIFY,
+    OP_HASH160,
+)
+from test_framework.test_framework import BitcoinTestFramework
+
+# Default DISCOURAGEMENT_THRESHOLD in net_processing.h.
+DISCOURAGEMENT_THRESHOLD = 100
+# Penalty applied when the relaying peer sends a DSTX whose masternode we
+# can't find in the deterministic MN list (and therefore can't verify).
+UNKNOWN_MN_SCORE = 1
+# Penalty applied when the DSTX itself is structurally bad / bad signature.
+INVALID_DSTX_SCORE = 10
+
+
+class P2PDSTXTest(BitcoinTestFramework):
+    def set_test_params(self):
+        self.num_nodes = 1
+        self.setup_clean_chain = True
+        self.extra_args = [["-debug=net", "-debug=coinjoin"]]
+
+    def make_dstx(self, nonce=0):
+        tx = CTransaction()
+        # The nonce flows into one of the prevouts so each DSTX has a distinct
+        # txid (and therefore is not deduped by dstxman.GetDSTX).
+        tx.vin = [CTxIn(COutPoint(hash=(nonce << 8) | (i + 1), n=0)) for i in range(3)]
+        p2pkh = CScript([
+            OP_DUP,
+            OP_HASH160,
+            b"\x01" * 20,
+            OP_EQUALVERIFY,
+            OP_CHECKSIG,
+        ])
+        # CoinJoin::IsDenominatedAmount requires a recognised denom; the
+        # smallest denom is 0.001 DASH + 0.0000001 fee == COIN//1000 + 1.
+        tx.vout = [CTxOut(nValue=COIN // 1000 + 1, scriptPubKey=p2pkh) for _ in tx.vin]
+        return CCoinJoinBroadcastTx(
+            tx=tx,
+            m_protxHash=1,
+            vchSig=b"\x01" * 96,
+            sigTime=int(time.time()),
+        )
+
+    def run_test(self):
+        node = self.nodes[0]
+        self.log.info("Leave IBD so unsolicited DSTX is processed")
+        self.generate(node, 1)
+
+        self.log.info("Unknown-MN DSTX => small (+%d) misbehavior penalty", UNKNOWN_MN_SCORE)
+        peer = node.add_p2p_connection(P2PInterface())
+        # Match the substring of the Misbehaving log that identifies the score
+        # jump of a fresh peer (0 -> 1) along with our message tag.
+        with node.assert_debug_log([
+            "Can't find masternode",
+            "Misbehaving",
+            "(0 -> {})".format(UNKNOWN_MN_SCORE),
+            "invalid dstx",
+        ]):
+            peer.send_and_ping(msg_dstx(self.make_dstx(nonce=1)))
+
+        self.log.info("Structurally invalid DSTX => stronger (+%d) misbehavior penalty", INVALID_DSTX_SCORE)
+        peer_invalid = node.add_p2p_connection(P2PInterface())
+        bad = self.make_dstx(nonce=2)
+        bad.tx.vout.pop()  # vin.size() != vout.size() trips IsValidStructure
+        with node.assert_debug_log([
+            "Invalid DSTX structure",
+            "Misbehaving",
+            "(0 -> {})".format(INVALID_DSTX_SCORE),
+            "invalid dstx",
+        ]):
+            peer_invalid.send_and_ping(msg_dstx(bad))
+
+        self.log.info("A peer flooding unknown-MN DSTXes is eventually discouraged")
+        peer_flood = node.add_p2p_connection(P2PInterface())
+        # +1 per unknown-MN DSTX, so DISCOURAGEMENT_THRESHOLD distinct DSTXes
+        # are enough to cross the threshold exactly once on this peer.
+        # send_message is fire-and-forget; the node disconnects once
+        # discouraged, so we wait on the threshold log line directly and then
+        # confirm the peer was dropped.
+        with node.assert_debug_log(["DISCOURAGE THRESHOLD EXCEEDED"], timeout=10):
+            for nonce in range(DISCOURAGEMENT_THRESHOLD):
+                # offset nonces to avoid txid clashes with earlier sends
+                peer_flood.send_message(msg_dstx(self.make_dstx(nonce=1000 + nonce)))
+        peer_flood.wait_for_disconnect()
+
+
+if __name__ == '__main__':
+    P2PDSTXTest().main()

test/functional/test_framework/messages.py

@@ -1835,6 +1835,53 @@ def __repr__(self):
         return "msg_tx(tx=%s)" % (repr(self.tx))
 
 
+class CCoinJoinBroadcastTx:
+    __slots__ = ("tx", "m_protxHash", "vchSig", "sigTime")
+
+    def __init__(self, tx=None, m_protxHash=0, vchSig=b"", sigTime=0):
+        self.tx = tx or CTransaction()
+        self.m_protxHash = m_protxHash
+        self.vchSig = vchSig
+        self.sigTime = sigTime
+
+    def deserialize(self, f):
+        self.tx = CTransaction()
+        self.tx.deserialize(f)
+        self.m_protxHash = deser_uint256(f)
+        self.vchSig = deser_string(f)
+        self.sigTime = struct.unpack("<q", f.read(8))[0]
+
+    def serialize(self):
+        r = b""
+        r += self.tx.serialize()
+        r += ser_uint256(self.m_protxHash)
+        r += ser_string(self.vchSig)
+        r += struct.pack("<q", self.sigTime)
+        return r
+
+    def __repr__(self):
+        return "CCoinJoinBroadcastTx(tx=%s m_protxHash=%064x)" \
+               % (repr(self.tx), self.m_protxHash)
+
+
+class msg_dstx:
+    __slots__ = ("dstx",)
+    msgtype = b"dstx"
+
+    def __init__(self, dstx=None):
+        self.dstx = dstx or CCoinJoinBroadcastTx()
+
+    def deserialize(self, f):
+        self.dstx = CCoinJoinBroadcastTx()
+        self.dstx.deserialize(f)
+
+    def serialize(self):
+        return self.dstx.serialize()
+
+    def __repr__(self):
+        return "msg_dstx(dstx=%s)" % (repr(self.dstx))
+
+
 class msg_block:
     __slots__ = ("block",)
     msgtype = b"block"

test/functional/test_runner.py

@@ -258,6 +258,7 @@
     'p2p_nobloomfilter_messages.py',
     'p2p_filter.py',
     'p2p_blocksonly.py',
+    'p2p_dstx.py',
     'rpc_setban.py --v1transport',
     'rpc_setban.py --v2transport',
     'mining_prioritisetransaction.py',