Add federated block-signing server implementation

src/federation.cpp

@@ -33,6 +33,7 @@
 
 #include <algorithm>
 #include <chrono>
+#include <functional>
 #include <memory>
 #include <map>
 #include <set>
@@ -242,9 +243,15 @@ static bool ReadFederationOpts() {
 //! and their associated public keys.
 static std::map<NodeId, CPubKey> g_fednode_conn;
 
+//! The current round, calculated from the beginning of (UNIX) time.  The actual
+//! number doesn't matter, so long as all peers agree on what it is.
+static int64_t g_roundno = 0;
+
 //! The current block proposal, which is announced by the leader at the
 //! beginning of their round.
 static CBlock g_block_proposal;
+//! A collection of ACK signatures for the block proposal of the current round.
+static std::map<CPubKey, std::vector<unsigned char>> g_block_acks;
 //! A collection of valid signatures for the proposed block of the current round.
 static std::map<CPubKey, std::vector<unsigned char>> g_block_sigs;
 
@@ -255,6 +262,9 @@ static std::map<CPubKey, std::vector<unsigned char>> g_block_sigs;
 namespace BlockSignMsgType {
 const unsigned char VER      = 0;
 const unsigned char VERACK   = 1;
+const unsigned char PROPOSAL = 2;
+const unsigned char ACK      = 3;
+const unsigned char SIG      = 4;
 } // namespace BlockSignMsgType
 
 // Checks if the specified peer is in the whitelist, and if so sends it a VER
@@ -336,6 +346,20 @@ static bool GenerateBlockProposal()
     return true;
 }
 
+struct AckBlockMsg
+{
+    uint256 m_block_hash;
+
+    AckBlockMsg(const uint256& hash) : m_block_hash(hash) { }
+    AckBlockMsg(const CBlock& block) : m_block_hash(block.GetHash()) { }
+
+    SERIALIZE_METHODS(AckBlockMsg, obj)
+    {
+        READWRITE(obj.m_block_hash);
+        READWRITE(BlockSignMsgType::ACK);
+    }
+};
+
 // Verifies that the proposed block is building off the tip and is a valid
 // block, then signs a relayable confirmation that we will sign this block and
 // accept it if the signing threshold is reached.
@@ -371,9 +395,100 @@ static bool AcceptBlockProposal()
         return false;
     }
 
+    // Produce the ack signature
+    std::vector<unsigned char> sig;
+#ifdef ENABLE_WALLET
+    CKey key;
+    std::shared_ptr<CWallet> pwallet = GetWalletForBlockSigning();
+    LegacyScriptPubKeyMan* spk_man = pwallet->GetLegacyScriptPubKeyMan();
+    if (spk_man && spk_man->GetKey(g_our_block_signing_key.GetID(), key)) {
+        CHashWriter ss(SER_GETHASH, 0);
+        ss << AckBlockMsg(block);
+        key.SignCompact(ss.GetHash(), sig);
+    }
+#endif // ENABLE_WALLET
+    if (sig.empty()) {
+        LogPrint(BCLog::FEDERATION, "Error: unable to produce block ACK signature.\n");
+        return false;
+    }
+
+    // Record our ack signature
+    g_block_acks[g_our_block_signing_key] = sig;
+    return true;
+}
+
+static void SendToEachPeer(std::function<bool(CNode* pnode)> func)
+{
+    LOCK(cs_block_signer);
+
+    for (auto& item : g_block_signers) {
+        BlockSigner& other = item.second;
+        // Attempt to send the message through each connection.  Log the
+        // connections for which sending fails.
+        std::vector<NodeId> nodes_to_remove;
+        for (const auto& nodeid : other.m_nodes) {
+            if (!g_context->connman->ForNode(nodeid, func)) {
+                nodes_to_remove.push_back(nodeid);
+            }
+        }
+        // Remove any connection in which the message couldn't be sent.
+        for (const auto& nodeid : nodes_to_remove) {
+            other.m_nodes.erase(nodeid);
+            g_fednode_conn.erase(nodeid);
+        }
+    }
+}
+
+static bool SendBlockProposalToPeer(CNode* pnode)
+{
+    // All paths return true because returning false is used to indicate that
+    // the connection has been dropped.
+
+    LOCK(cs_block_signer);
+
+    if (!g_block_acks.count(g_our_block_signing_key)) {
+        LogPrintf("Error: %s called for a block that we haven't ACK'd.  Did the round end while we were generating the block?  Otherwise this should never happen.\n", __func__);
+        return true;
+    }
+
+    g_context->connman->PushMessage(pnode, CNetMsgMaker(pnode->GetCommonVersion()).Make(
+        NetMsgType::BLOCKSIGN,
+        BlockSignMsgType::PROPOSAL,
+        g_our_block_signing_key,
+        g_block_proposal,
+        g_block_acks[g_our_block_signing_key]));
+    return true;
+}
+
+static void SendBlockProposal()
+{
+    SendToEachPeer(SendBlockProposalToPeer);
+}
+
+static bool SendAckToPeer(CNode* pnode)
+{
+    // All paths return true because returning false is used to indicate that
+    // the connection has been dropped.
+
+    LOCK(cs_block_signer);
+
+    if (g_block_acks.count(g_our_block_signing_key)) {
+        g_context->connman->PushMessage(pnode, CNetMsgMaker(pnode->GetCommonVersion()).Make(
+            NetMsgType::BLOCKSIGN,
+            BlockSignMsgType::ACK,
+            g_our_block_signing_key,
+            g_block_proposal.GetHash(),
+            g_block_acks[g_our_block_signing_key]));
+    }
+
     return true;
 }
 
+static void SendAck()
+{
+    SendToEachPeer(SendAckToPeer);
+}
+
 static bool SignBlock()
 {
     LOCK(cs_block_signer);
@@ -382,6 +497,13 @@ static bool SignBlock()
         return true;
     }
 
+    // Check that we have ACK'd this block.  This serves as a cached check that
+    // the block proposal validates.
+    if (!g_block_acks.count(g_our_block_signing_key)) {
+        LogPrint(BCLog::FEDERATION, "%s called for a block proposal that we haven't ACK'd (yet).  This should never happen!\n", __func__);
+        return false;
+    }
+
     // Generate the block signature
     std::vector<unsigned char> sig;
 #ifdef ENABLE_WALLET
@@ -411,6 +533,32 @@ static bool SignBlock()
     return true;
 }
 
+static bool SendBlockSigToPeer(CNode* pnode)
+{
+    // All paths return true because returning false is used to indicate that
+    // the connection has been dropped.
+
+    LOCK(cs_block_signer);
+
+    if (!g_block_sigs.count(g_our_block_signing_key)) {
+        LogPrint(BCLog::FEDERATION, "%s called before we signed the current block proposal.  Did the current round end while we were sending our signature?  Otherwise this should never happen.\n", __func__);
+        return true;
+    }
+
+    g_context->connman->PushMessage(pnode, CNetMsgMaker(pnode->GetCommonVersion()).Make(
+        NetMsgType::BLOCKSIGN,
+        BlockSignMsgType::SIG,
+        g_our_block_signing_key,
+        g_block_proposal.GetHash(),
+        g_block_sigs[g_our_block_signing_key]));
+    return true;
+}
+
+static void SendBlockSig()
+{
+    SendToEachPeer(SendBlockSigToPeer);
+}
+
 static bool SubmitBlock()
 {
     LOCK(cs_block_signer);
@@ -493,30 +641,39 @@ void ThreadBlockSigner(NodeContext& node)
             using std::swap;
             CBlock empty;
             swap(g_block_proposal, empty);
+            g_block_acks.clear();
             g_block_sigs.clear();
         }
 
-        if (!GenerateBlockProposal()) {
-            LogPrint(BCLog::FEDERATION, "%s: Unable to generate block proposal.  Terminating round early.\n", __func__);
+        int64_t sec_since_epoch = nextblocktime.time_since_epoch().count();
+        g_roundno = sec_since_epoch / 5;
+        int turn = g_roundno % g_block_signing_keys.size();
+        if (turn != g_our_block_signing_key_index) {
+            // Not our round
+            LogPrint(BCLog::FEDERATION, "Not our round; skipping\n");
             continue;
         }
 
-        if (!AcceptBlockProposal()) {
-            LogPrint(BCLog::FEDERATION, "%s: Unable to validate our block proposal.  Terminating round early.\n", __func__);
-            continue;
+        // Delay 1 second to make sure our peers have transitioned to the
+        // new round before sending our block proposal.
+        std::chrono::steady_clock::time_point delay =
+            nextblocktime + std::chrono::seconds{1};
+        if (!g_thread_interrupt.sleep_until(delay)) {
+            // Thread interrupted.  Shutdown in progress?
+            break;
         }
 
-        if (!SignBlock()) {
-            LogPrint(BCLog::FEDERATION, "%s: Unable to sign block.  Terminating round early.\n", __func__);
+        if (!GenerateBlockProposal()) {
+            LogPrint(BCLog::FEDERATION, "%s: Unable to generate block proposal.  Terminating round early.\n", __func__);
             continue;
         }
 
-        if (!SubmitBlock()) {
-            LogPrint(BCLog::FEDERATION, "%s: Unable to submit block.  Terminating round early.\n", __func__);
+        if (!AcceptBlockProposal()) {
+            LogPrint(BCLog::FEDERATION, "%s: Unable to validate our block proposal.  Terminating round early.\n", __func__);
             continue;
         }
 
-        LogPrint(BCLog::FEDERATION, "Generated new block: %s\n", g_block_proposal.GetHash().GetHex());
+        SendBlockProposal();
     }
 }
 
@@ -575,6 +732,108 @@ void HandleBlockSignMessage(CNode& from, CDataStream& msg, std::chrono::microsec
         }
         break;
 
+        case BlockSignMsgType::PROPOSAL:
+        {
+            LOCK(cs_block_signer);
+
+            CPubKey other;
+            msg >> other;
+            if (other != g_block_signing_keys[g_roundno % g_block_signing_keys.size()]) {
+                std::string pkstr(HexStr(CDataStream(SER_NETWORK, CLIENT_VERSION) << other), 2);
+                LogPrint(BCLog::FEDERATION, "Received out-of-turn block proposal from peer %s.  Ignoring.\n", pkstr);
+                break;
+            }
+
+            msg >> g_block_proposal;
+
+            // FIXME: read the ACK signature, validate it, and store it
+        }
+        AcceptBlockProposal();
+        SendAck();
+        break;
+
+        case BlockSignMsgType::ACK:
+        {
+            LOCK(cs_block_signer);
+
+            CPubKey other;
+            msg >> other;
+            if (std::find(g_block_signing_keys.begin(), g_block_signing_keys.end(), other) == g_block_signing_keys.end()) {
+                LogPrint(BCLog::FEDERATION, "Error: received BlockSignMsgType::ACK message with unknown pubkey.\n");
+                break;
+            }
+
+            uint256 block_hash;
+            msg >> block_hash;
+            if (block_hash != g_block_proposal.GetHash()) {
+                LogPrint(BCLog::FEDERATION, "Error: received BlockSignMsgType::ACK message for block which does not match current proposal. (%s != %s)\n", block_hash.GetHex(), g_block_proposal.GetHash().GetHex());
+                break;
+            }
+            CHashWriter ss(SER_GETHASH, 0);
+            ss << AckBlockMsg(block_hash);
+
+            std::vector<unsigned char> sig;
+            msg >> sig;
+
+            CPubKey pk;
+            if (!pk.RecoverCompact(ss.GetHash(), sig)) {
+                LogPrint(BCLog::FEDERATION, "Error: unable to recover pubkey; is this really a ECDSA signature?\n");
+                break;
+            }
+            if (pk != other) {
+                LogPrint(BCLog::FEDERATION, "Error: recovered pubkey does not match expected signing key.\n");
+                break;
+            }
+
+            g_block_acks[pk] = sig;
+
+            auto missing_acks = g_block_signing_keys.size() - g_block_acks.size();
+            if (missing_acks > (g_block_signing_keys.size() / 3)) {
+                LogPrint(BCLog::FEDERATION, "Not enough keys to advance to signing round. (%d > %d)\n", missing_acks, g_block_signing_keys.size() / 3);
+                break;
+            }
+#if 0
+            if (g_block_sigs.count(g_our_block_signing_key)) {
+                LogPrint(BCLog::FEDERATION, "Already sent our block sig; won't send again.\n");
+                break;
+            }
+#endif
+        }
+        if (!SignBlock()) {
+            LogPrintf("Error: failed to create block signature.\n");
+            break;
+        }
+        SendBlockSig();
+        break;
+
+        case BlockSignMsgType::SIG:
+        {
+            LOCK(cs_block_signer);
+
+            CPubKey other;
+            msg >> other;
+            if (std::find(g_block_signing_keys.begin(), g_block_signing_keys.end(), other) == g_block_signing_keys.end()) {
+                LogPrint(BCLog::FEDERATION, "Error: received BlockSignMsgType::SIG message with unknown pubkey.\n");
+                break;
+            }
+
+            uint256 block_hash;
+            msg >> block_hash;
+            if (block_hash != g_block_proposal.GetHash()) {
+                LogPrint(BCLog::FEDERATION, "Error: received BlockSignMsgType::SIG message for block which does not match current proposal. (%s != %s)\n", block_hash.GetHex(), g_block_proposal.GetHash().GetHex());
+                break;
+            }
+
+            std::vector<unsigned char> sig;
+            msg >> sig;
+
+            // FIXME: check if signature is valid!
+
+            g_block_sigs[other] = sig;
+        }
+        SubmitBlock();
+        break;
+
         default:
             LogPrint(BCLog::FEDERATION, "Received an unrecognized blocksign message (code=%d) from peer=%d.  Ignoring.\n", code, from.GetId());
         break;