Allow re-signing of IS locks when performing retroactive signing (das…

…hpay#3219)

* Implement re-signing of InstantSend inputs when TXs come in via blocks

* Use GetAdjustedTime instead of GetTimeMillis in CSigSharesManager

This allows use of mocktime in tests.

* Expose verifiedProRegTxHash in getpeerinfo and implement wait_for_mnauth

* Allow to wait for IS and CL to NOT happen

* Bump timeout for wait_for_instantlock

* Implement tests for retroactive signing of IS and CLs

* Add wait_for_tx function to IonTestFramework

* Add -whitelist=127.0.0.1 to node0

* Use node3 for isolated block generation

* Don't test for non-receival of TXs on node4/node5

Signed-off-by: cevap <dev@i2pmail.org>

src/llmq/quorums_instantsend.cpp

@@ -371,7 +371,7 @@ void CInstantSendManager::InterruptWorkerThread()
     workInterrupt();
 }
 
-bool CInstantSendManager::ProcessTx(const CTransaction& tx, const Consensus::Params& params)
+bool CInstantSendManager::ProcessTx(const CTransaction& tx, bool allowReSigning, const Consensus::Params& params)
 {
     if (!IsInstantSendEnabled()) {
         return true;
@@ -441,7 +441,7 @@ bool CInstantSendManager::ProcessTx(const CTransaction& tx, const Consensus::Par
             return false;
         }
     }
-    if (alreadyVotedCount == ids.size()) {
+    if (!allowReSigning && alreadyVotedCount == ids.size()) {
         LogPrint(BCLog::INSTANTSEND, "CInstantSendManager::%s -- txid=%s: already voted on all inputs, bailing out\n", __func__,
                  tx.GetHash().ToString());
         return true;
@@ -454,9 +454,9 @@ bool CInstantSendManager::ProcessTx(const CTransaction& tx, const Consensus::Par
         auto& in = tx.vin[i];
         auto& id = ids[i];
         inputRequestIds.emplace(id);
-        LogPrint(BCLog::INSTANTSEND, "CInstantSendManager::%s -- txid=%s: trying to vote on input %s with id %s\n", __func__,
-                 tx.GetHash().ToString(), in.prevout.ToStringShort(), id.ToString());
-        if (quorumSigningManager->AsyncSignIfMember(llmqType, id, tx.GetHash())) {
+        LogPrint(BCLog::INSTANTSEND, "CInstantSendManager::%s -- txid=%s: trying to vote on input %s with id %s. allowReSigning=%d\n", __func__,
+                 tx.GetHash().ToString(), in.prevout.ToStringShort(), id.ToString(), allowReSigning);
+        if (quorumSigningManager->AsyncSignIfMember(llmqType, id, tx.GetHash(), allowReSigning)) {
             LogPrint(BCLog::INSTANTSEND, "CInstantSendManager::%s -- txid=%s: voted on input %s with id %s\n", __func__,
                       tx.GetHash().ToString(), in.prevout.ToStringShort(), id.ToString());
         }
@@ -961,7 +961,7 @@ void CInstantSendManager::UpdateWalletTransaction(const CTransactionRef& tx, con
     mempool.AddTransactionsUpdated(1);
 }
 
-void CInstantSendManager::ProcessNewTransaction(const CTransactionRef& tx, const CBlockIndex* pindex)
+void CInstantSendManager::ProcessNewTransaction(const CTransactionRef& tx, const CBlockIndex* pindex, bool allowReSigning)
 {
     if (!IsInstantSendEnabled()) {
         return;
@@ -989,7 +989,7 @@ void CInstantSendManager::ProcessNewTransaction(const CTransactionRef& tx, const
 
     bool chainlocked = pindex && chainLocksHandler->HasChainLock(pindex->nHeight, pindex->GetBlockHash());
     if (islockHash.IsNull() && !chainlocked) {
-        ProcessTx(*tx, Params().GetConsensus());
+        ProcessTx(*tx, allowReSigning, Params().GetConsensus());
     }
 
     LOCK(cs);
@@ -1004,7 +1004,7 @@ void CInstantSendManager::ProcessNewTransaction(const CTransactionRef& tx, const
 
 void CInstantSendManager::TransactionAddedToMempool(const CTransactionRef& tx)
 {
-    ProcessNewTransaction(tx, nullptr);
+    ProcessNewTransaction(tx, nullptr, false);
 }
 
 void CInstantSendManager::BlockConnected(const std::shared_ptr<const CBlock>& pblock, const CBlockIndex* pindex, const std::vector<CTransactionRef>& vtxConflicted)
@@ -1021,7 +1021,7 @@ void CInstantSendManager::BlockConnected(const std::shared_ptr<const CBlock>& pb
     }
 
     for (const auto& tx : pblock->vtx) {
-        ProcessNewTransaction(tx, pindex);
+        ProcessNewTransaction(tx, pindex, true);
     }
 }
 
@@ -1400,7 +1400,7 @@ bool CInstantSendManager::ProcessPendingRetryLockTxs()
                      tx->GetHash().ToString());
         }
 
-        ProcessTx(*tx, Params().GetConsensus());
+        ProcessTx(*tx, false, Params().GetConsensus());
         retryCount++;
     }
 

src/llmq/quorums_instantsend.h

@@ -120,7 +120,7 @@ class CInstantSendManager : public CRecoveredSigsListener
     void InterruptWorkerThread();
 
 public:
-    bool ProcessTx(const CTransaction& tx, const Consensus::Params& params);
+    bool ProcessTx(const CTransaction& tx, bool allowReSigning, const Consensus::Params& params);
     bool CheckCanLock(const CTransaction& tx, bool printDebug, const Consensus::Params& params);
     bool CheckCanLock(const COutPoint& outpoint, bool printDebug, const uint256& txHash, CAmount* retValue, const Consensus::Params& params);
     bool IsLocked(const uint256& txHash);
@@ -141,7 +141,7 @@ class CInstantSendManager : public CRecoveredSigsListener
     void ProcessInstantSendLock(NodeId from, const uint256& hash, const CInstantSendLock& islock);
     void UpdateWalletTransaction(const CTransactionRef& tx, const CInstantSendLock& islock);
 
-    void ProcessNewTransaction(const CTransactionRef& tx, const CBlockIndex* pindex);
+    void ProcessNewTransaction(const CTransactionRef& tx, const CBlockIndex* pindex, bool allowReSigning);
     void TransactionAddedToMempool(const CTransactionRef& tx);
     void BlockConnected(const std::shared_ptr<const CBlock>& pblock, const CBlockIndex* pindex, const std::vector<CTransactionRef>& vtxConflicted);
     void BlockDisconnected(const std::shared_ptr<const CBlock>& pblock, const CBlockIndex* pindexDisconnected);

src/llmq/quorums_signing.cpp

@@ -762,7 +762,7 @@ void CSigningManager::UnregisterRecoveredSigsListener(CRecoveredSigsListener* l)
     recoveredSigsListeners.erase(itRem, recoveredSigsListeners.end());
 }
 
-bool CSigningManager::AsyncSignIfMember(Consensus::LLMQType llmqType, const uint256& id, const uint256& msgHash)
+bool CSigningManager::AsyncSignIfMember(Consensus::LLMQType llmqType, const uint256& id, const uint256& msgHash, bool allowReSign)
 {
     auto& params = Params().GetConsensus().llmqs.at(llmqType);
 
@@ -773,24 +773,31 @@ bool CSigningManager::AsyncSignIfMember(Consensus::LLMQType llmqType, const uint
     {
         LOCK(cs);
 
-        if (db.HasVotedOnId(llmqType, id)) {
+        bool hasVoted = db.HasVotedOnId(llmqType, id);
+        if (hasVoted) {
             uint256 prevMsgHash;
             db.GetVoteForId(llmqType, id, prevMsgHash);
             if (msgHash != prevMsgHash) {
                 LogPrintf("CSigningManager::%s -- already voted for id=%s and msgHash=%s. Not voting on conflicting msgHash=%s\n", __func__,
                         id.ToString(), prevMsgHash.ToString(), msgHash.ToString());
+                return false;
+            } else if (allowReSign) {
+                LogPrint(BCLog::LLMQ, "CSigningManager::%s -- already voted for id=%s and msgHash=%s. Resigning!\n", __func__,
+                         id.ToString(), prevMsgHash.ToString());
             } else {
                 LogPrint(BCLog::LLMQ, "CSigningManager::%s -- already voted for id=%s and msgHash=%s. Not voting again.\n", __func__,
                           id.ToString(), prevMsgHash.ToString());
+                return false;
             }
-            return false;
         }
 
         if (db.HasRecoveredSigForId(llmqType, id)) {
             // no need to sign it if we already have a recovered sig
             return true;
         }
-        db.WriteVoteForId(llmqType, id, msgHash);
+        if (!hasVoted) {
+            db.WriteVoteForId(llmqType, id, msgHash);
+        }
     }
 
     int tipHeight;
@@ -814,6 +821,10 @@ bool CSigningManager::AsyncSignIfMember(Consensus::LLMQType llmqType, const uint
         return false;
     }
 
+    if (allowReSign) {
+        // make us re-announce all known shares (other nodes might have run into a timeout)
+        quorumSigSharesManager->ForceReAnnouncement(quorum, llmqType, id, msgHash);
+    }
     quorumSigSharesManager->AsyncSign(quorum, id, msgHash);
 
     return true;

src/llmq/quorums_signing.h

@@ -170,7 +170,7 @@ class CSigningManager
     void RegisterRecoveredSigsListener(CRecoveredSigsListener* l);
     void UnregisterRecoveredSigsListener(CRecoveredSigsListener* l);
 
-    bool AsyncSignIfMember(Consensus::LLMQType llmqType, const uint256& id, const uint256& msgHash);
+    bool AsyncSignIfMember(Consensus::LLMQType llmqType, const uint256& id, const uint256& msgHash, bool allowReSign = false);
     bool HasRecoveredSig(Consensus::LLMQType llmqType, const uint256& id, const uint256& msgHash);
     bool HasRecoveredSigForId(Consensus::LLMQType llmqType, const uint256& id);
     bool HasRecoveredSigForSession(const uint256& signHash);

src/llmq/quorums_signing_shares.cpp

@@ -82,6 +82,13 @@ void CSigSharesInv::Set(uint16_t quorumMember, bool v)
     inv[quorumMember] = v;
 }
 
+void CSigSharesInv::SetAll(bool v)
+{
+    for (size_t i = 0; i < inv.size(); i++) {
+        inv[i] = v;
+    }
+}
+
 std::string CBatchedSigShares::ToInvString() const
 {
     CSigSharesInv inv;
@@ -678,7 +685,7 @@ void CSigSharesManager::ProcessSigShare(NodeId nodeId, const CSigShare& sigShare
         sigSharesToAnnounce.Add(sigShare.GetKey(), true);
 
         // Update the time we've seen the last sigShare
-        timeSeenForSessions[sigShare.GetSignHash()] = GetTimeMillis();
+        timeSeenForSessions[sigShare.GetSignHash()] = GetAdjustedTime();
 
         if (!quorumNodes.empty()) {
             // don't announce and wait for other nodes to request this share and directly send it to them
@@ -777,7 +784,7 @@ void CSigSharesManager::CollectSigSharesToRequest(std::unordered_map<NodeId, std
 {
     AssertLockHeld(cs);
 
-    int64_t now = GetTimeMillis();
+    int64_t now = GetAdjustedTime();
     const size_t maxRequestsForNode = 32;
 
     // avoid requesting from same nodes all the time
@@ -1143,8 +1150,8 @@ CSigShare CSigSharesManager::RebuildSigShare(const CSigSharesNodeState::SessionI
 
 void CSigSharesManager::Cleanup()
 {
-    int64_t now = GetTimeMillis();
-    if (now - lastCleanupTime < 5000) {
+    int64_t now = GetAdjustedTime();
+    if (now - lastCleanupTime < 5) {
         return;
     }
 
@@ -1265,7 +1272,7 @@ void CSigSharesManager::Cleanup()
         nodeStates.erase(nodeId);
     }
 
-    lastCleanupTime = GetTimeMillis();
+    lastCleanupTime = GetAdjustedTime();
 }
 
 void CSigSharesManager::RemoveSigSharesForSession(const uint256& signHash)
@@ -1426,6 +1433,31 @@ void CSigSharesManager::Sign(const CQuorumCPtr& quorum, const uint256& id, const
     ProcessSigShare(-1, sigShare, *g_connman, quorum);
 }
 
+// causes all known sigShares to be re-announced
+void CSigSharesManager::ForceReAnnouncement(const CQuorumCPtr& quorum, Consensus::LLMQType llmqType, const uint256& id, const uint256& msgHash)
+{
+    LOCK(cs);
+    auto signHash = CLLMQUtils::BuildSignHash(llmqType, quorum->qc.quorumHash, id, msgHash);
+    auto sigs = sigShares.GetAllForSignHash(signHash);
+    if (sigs) {
+        for (auto& p : *sigs) {
+            // re-announce every sigshare to every node
+            sigSharesToAnnounce.Add(std::make_pair(signHash, p.first), true);
+        }
+    }
+    for (auto& p : nodeStates) {
+        CSigSharesNodeState& nodeState = p.second;
+        auto session = nodeState.GetSessionBySignHash(signHash);
+        if (!session) {
+            continue;
+        }
+        // pretend that the other node doesn't know about any shares so that we re-announce everything
+        session->knows.SetAll(false);
+        // we need to use a new session id as we don't know if the other node has run into a timeout already
+        session->sendSessionId = (uint32_t)-1;
+    }
+}
+
 void CSigSharesManager::HandleNewRecoveredSig(const llmq::CRecoveredSig& recoveredSig)
 {
     LOCK(cs);

src/llmq/quorums_signing_shares.h

@@ -104,6 +104,7 @@ class CSigSharesInv
     void Init(size_t size);
     bool IsSet(uint16_t quorumMember) const;
     void Set(uint16_t quorumMember, bool v);
+    void SetAll(bool v);
     void Merge(const CSigSharesInv& inv2);
 
     size_t CountSet() const;
@@ -328,8 +329,8 @@ class CSigSharesNodeState
 
 class CSigSharesManager : public CRecoveredSigsListener
 {
-    static const int64_t SESSION_NEW_SHARES_TIMEOUT = 60 * 1000;
-    static const int64_t SIG_SHARE_REQUEST_TIMEOUT = 5 * 1000;
+    static const int64_t SESSION_NEW_SHARES_TIMEOUT = 60;
+    static const int64_t SIG_SHARE_REQUEST_TIMEOUT = 5;
 
     // we try to keep total message size below 10k
     const size_t MAX_MSGS_CNT_QSIGSESANN = 100;
@@ -376,6 +377,7 @@ class CSigSharesManager : public CRecoveredSigsListener
 
     void AsyncSign(const CQuorumCPtr& quorum, const uint256& id, const uint256& msgHash);
     void Sign(const CQuorumCPtr& quorum, const uint256& id, const uint256& msgHash);
+    void ForceReAnnouncement(const CQuorumCPtr& quorum, Consensus::LLMQType llmqType, const uint256& id, const uint256& msgHash);
 
     void HandleNewRecoveredSig(const CRecoveredSig& recoveredSig);
 

src/net.cpp

@@ -717,6 +717,11 @@ void CNode::copyStats(CNodeStats &stats)
     // Leave string empty if addrLocal invalid (not filled in yet)
     CService addrLocalUnlocked = GetAddrLocal();
     stats.addrLocal = addrLocalUnlocked.IsValid() ? addrLocalUnlocked.ToString() : "";
+
+    {
+        LOCK(cs_mnauth);
+        X(verifiedProRegTxHash);
+    }
 }
 #undef X
 

src/net.h

@@ -699,6 +699,8 @@ class CNodeStats
     CAddress addr;
     // Bind address of our side of the connection
     CAddress addrBind;
+    // In case this is a verified MN, this value is the proTx of the MN
+    uint256 verifiedProRegTxHash;
 };
 
 

src/rpc/net.cpp

@@ -80,6 +80,9 @@ UniValue getpeerinfo(const JSONRPCRequest& request)
             "    \"addrbind\":\"ip:port\",    (string) Bind address of the connection to the peer\n"
             "    \"addrlocal\":\"ip:port\",   (string) Local address as reported by the peer\n"
             "    \"services\":\"xxxxxxxxxxxxxxxx\",   (string) The services offered\n"
+            "    \"verified_proregtx_hash\": h, (hex) Only present when the peer is a masternode and succesfully\n"
+            "                               autheticated via MNAUTH. In this case, this field contains the\n"
+            "                               protx hash of the masternode\n"
             "    \"relaytxes\":true|false,    (boolean) Whether peer has asked us to relay transactions to it\n"
             "    \"lastsend\": ttt,           (numeric) The time in seconds since epoch (Jan 1 1970 GMT) of the last send\n"
             "    \"lastrecv\": ttt,           (numeric) The time in seconds since epoch (Jan 1 1970 GMT) of the last receive\n"
@@ -138,6 +141,9 @@ UniValue getpeerinfo(const JSONRPCRequest& request)
         if (stats.addrBind.IsValid())
             obj.push_back(Pair("addrbind", stats.addrBind.ToString()));
         obj.push_back(Pair("services", strprintf("%016x", stats.nServices)));
+        if (!stats.verifiedProRegTxHash.IsNull()) {
+            obj.push_back(Pair("verified_proregtx_hash", stats.verifiedProRegTxHash.ToString()));
+        }
         obj.push_back(Pair("relaytxes", stats.fRelayTxes));
         obj.push_back(Pair("lastsend", stats.nLastSend));
         obj.push_back(Pair("lastrecv", stats.nLastRecv));

test/functional/llmq-chainlocks.py

@@ -129,7 +129,7 @@ def run_test(self):
         # for the mined TXs, which will then allow the network to create a CLSIG
         self.log.info("Reenable network on first node and wait for chainlock")
         reconnect_isolated_node(self.nodes[0], 1)
-        self.wait_for_chainlocked_block(self.nodes[0], self.nodes[0].getbestblockhash(), 30)
+        self.wait_for_chainlocked_block(self.nodes[0], self.nodes[0].getbestblockhash(), timeout=30)
 
     def create_chained_txs(self, node, amount):
         txid = node.sendtoaddress(node.getnewaddress(), amount)

test/functional/test_framework/test_framework.py

@@ -698,26 +698,37 @@ def create_raw_tx(self, node_from, node_to, amount, min_inputs, max_inputs):
         ret = {**decoded, **ret}
         return ret
 
-    def wait_for_instantlock(self, txid, node):
+    def wait_for_tx(self, txid, node, expected=True, timeout=15):
+        def check_tx():
+            try:
+                return node.getrawtransaction(txid)
+            except:
+                return False
+        if wait_until(check_tx, timeout=timeout, sleep=0.5, do_assert=expected) and not expected:
+            raise AssertionError("waiting unexpectedly succeeded")
+
+    def wait_for_instantlock(self, txid, node, expected=True, timeout=15):
         def check_instantlock():
             try:
                 return node.getrawtransaction(txid, True)["instantlock"]
             except:
                 return False
-        wait_until(check_instantlock, timeout=10, sleep=0.5)
+        if wait_until(check_instantlock, timeout=timeout, sleep=0.5, do_assert=expected) and not expected:
+            raise AssertionError("waiting unexpectedly succeeded")
 
-    def wait_for_chainlocked_block(self, node, block_hash, timeout=15):
+    def wait_for_chainlocked_block(self, node, block_hash, expected=True, timeout=15):
         def check_chainlocked_block():
             try:
                 block = node.getblock(block_hash)
                 return block["confirmations"] > 0 and block["chainlock"]
             except:
                 return False
-        wait_until(check_chainlocked_block, timeout=timeout, sleep=0.1)
+        if wait_until(check_chainlocked_block, timeout=timeout, sleep=0.1, do_assert=expected) and not expected:
+            raise AssertionError("waiting unexpectedly succeeded")
 
     def wait_for_chainlocked_block_all_nodes(self, block_hash, timeout=15):
         for node in self.nodes:
-            self.wait_for_chainlocked_block(node, block_hash, timeout)
+            self.wait_for_chainlocked_block(node, block_hash, timeout=timeout)
 
     def wait_for_best_chainlock(self, node, block_hash, timeout=15):
         wait_until(lambda: node.getbestchainlock()["blockhash"] == block_hash, timeout=timeout, sleep=0.1)
@@ -845,6 +856,17 @@ def mine_quorum(self, expected_contributions=5, expected_complaints=0, expected_
 
         return new_quorum
 
+    def wait_for_mnauth(self, node, count, timeout=10):
+        def test():
+            pi = node.getpeerinfo()
+            c = 0
+            for p in pi:
+                if "verified_proregtx_hash" in p and p["verified_proregtx_hash"] != "":
+                    c += 1
+            return c >= count
+        wait_until(test, timeout=timeout)
+
+
 class ComparisonTestFramework(BitcoinTestFramework):
     """Test framework for doing p2p comparison testing