Implement new way of concentrated signature recovery

Instead of propagating all sig shares to all LLMQ members, this will now
make all members send their individual sig share to a single member, which
is then responsible for the recovery and propagation of the recovered
signature. This process is repeated by all members every second for another
target/recovering member, until a recovered signature appears.
This commit is contained in:
Alexander Block 2020-03-17 07:08:37 +01:00
parent 45064d8dc9
commit b212f21c15
6 changed files with 209 additions and 8 deletions

View File

@ -173,6 +173,7 @@ static Consensus::LLMQParams llmq_test = {
.signingActiveQuorumCount = 2, // just a few ones to allow easier testing
.keepOldConnections = 3,
.recoveryMembers = 3,
};
// this one is for devnets only
@ -192,6 +193,7 @@ static Consensus::LLMQParams llmq_devnet = {
.signingActiveQuorumCount = 3, // just a few ones to allow easier testing
.keepOldConnections = 4,
.recoveryMembers = 6,
};
static Consensus::LLMQParams llmq50_60 = {
@ -210,6 +212,7 @@ static Consensus::LLMQParams llmq50_60 = {
.signingActiveQuorumCount = 24, // a full day worth of LLMQs
.keepOldConnections = 25,
.recoveryMembers = 25,
};
static Consensus::LLMQParams llmq400_60 = {
@ -228,6 +231,7 @@ static Consensus::LLMQParams llmq400_60 = {
.signingActiveQuorumCount = 4, // two days worth of LLMQs
.keepOldConnections = 5,
.recoveryMembers = 100,
};
// Used for deployment and min-proto-version signalling, so it needs a higher threshold
@ -247,6 +251,7 @@ static Consensus::LLMQParams llmq400_85 = {
.signingActiveQuorumCount = 4, // four days worth of LLMQs
.keepOldConnections = 5,
.recoveryMembers = 100,
};

View File

@ -114,6 +114,9 @@ struct LLMQParams {
// Used for inter-quorum communication. This is the number of quorums for which we should keep old connections. This
// should be at least one more then the active quorums set.
int keepOldConnections;
// How many members should we try to send all sigShares to before we give up.
int recoveryMembers;
};
/**

View File

@ -11,6 +11,7 @@
#include <init.h>
#include <net_processing.h>
#include <netmessagemaker.h>
#include <spork.h>
#include <validation.h>
#include <cxxtimer.hpp>
@ -236,6 +237,23 @@ void CSigSharesManager::ProcessMessage(CNode* pfrom, const std::string& strComma
return;
}
if (sporkManager.IsSporkActive(SPORK_21_QUORUM_ALL_CONNECTED)) {
if (strCommand == NetMsgType::QSIGSHARE) {
std::vector<CSigShare> sigShares;
vRecv >> sigShares;
if (sigShares.size() > MAX_MSGS_SIG_SHARES) {
LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- too many sigs in QSIGSHARE message. cnt=%d, max=%d, node=%d\n", __func__, sigShares.size(), MAX_MSGS_SIG_SHARES, pfrom->GetId());
BanNode(pfrom->GetId());
return;
}
for (auto& sigShare : sigShares) {
ProcessMessageSigShare(pfrom->GetId(), sigShare, connman);
}
}
}
if (strCommand == NetMsgType::QSIGSESANN) {
std::vector<CSigSesAnn> msgs;
vRecv >> msgs;
@ -465,6 +483,57 @@ bool CSigSharesManager::ProcessMessageBatchedSigShares(CNode* pfrom, const CBatc
return true;
}
void CSigSharesManager::ProcessMessageSigShare(NodeId fromId, const CSigShare& sigShare, CConnman& connman)
{
auto quorum = quorumManager->GetQuorum(sigShare.llmqType, sigShare.quorumHash);
if (!quorum) {
return;
}
if (!CLLMQUtils::IsQuorumActive(sigShare.llmqType, quorum->qc.quorumHash)) {
// quorum is too old
return;
}
if (!quorum->IsMember(activeMasternodeInfo.proTxHash)) {
// we're not a member so we can't verify it (we actually shouldn't have received it)
return;
}
if (quorum->quorumVvec == nullptr) {
// TODO we should allow to ask other nodes for the quorum vvec if we missed it in the DKG
LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- we don't have the quorum vvec for %s, no verification possible. node=%d\n", __func__,
quorum->qc.quorumHash.ToString(), fromId);
return;
}
if (sigShare.quorumMember >= quorum->members.size()) {
LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- quorumMember out of bounds\n", __func__);
BanNode(fromId);
return;
}
if (!quorum->qc.validMembers[sigShare.quorumMember]) {
LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- quorumMember not valid\n", __func__);
BanNode(fromId);
return;
}
{
LOCK(cs);
if (sigShares.Has(sigShare.GetKey())) {
return;
}
if (quorumSigningManager->HasRecoveredSigForId((Consensus::LLMQType)sigShare.llmqType, sigShare.id)) {
return;
}
auto& nodeState = nodeStates[fromId];
nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare);
}
LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- signHash=%s, id=%s, msgHash=%s, member=%d, node=%d\n", __func__,
sigShare.GetSignHash().ToString(), sigShare.id.ToString(), sigShare.msgHash.ToString(), sigShare.quorumMember, fromId);
}
bool CSigSharesManager::PreVerifyBatchedSigShares(NodeId nodeId, const CSigSharesNodeState::SessionInfo& session, const CBatchedSigShares& batchedSigShares, bool& retBan)
{
retBan = false;
@ -668,8 +737,10 @@ void CSigSharesManager::ProcessSigShare(NodeId nodeId, const CSigShare& sigShare
// prepare node set for direct-push in case this is our sig share
std::set<NodeId> quorumNodes;
if (sigShare.quorumMember == quorum->GetMemberIndex(activeMasternodeInfo.proTxHash)) {
quorumNodes = connman.GetMasternodeQuorumNodes((Consensus::LLMQType) sigShare.llmqType, sigShare.quorumHash);
if (!sporkManager.IsSporkActive(SPORK_21_QUORUM_ALL_CONNECTED)) {
if (sigShare.quorumMember == quorum->GetMemberIndex(activeMasternodeInfo.proTxHash)) {
quorumNodes = connman.GetMasternodeQuorumNodes((Consensus::LLMQType) sigShare.llmqType, sigShare.quorumHash);
}
}
if (quorumSigningManager->HasRecoveredSigForId(llmqType, sigShare.id)) {
@ -780,6 +851,21 @@ void CSigSharesManager::TryRecoverSig(const CQuorumCPtr& quorum, const uint256&
quorumSigningManager->ProcessRecoveredSig(-1, rs, quorum, connman);
}
CDeterministicMNCPtr CSigSharesManager::SelectMemberForRecovery(const CQuorumCPtr& quorum, const uint256 &id, int attempt)
{
assert(attempt < quorum->members.size());
std::vector<std::pair<uint256, CDeterministicMNCPtr>> v;
v.reserve(quorum->members.size());
for (size_t i = 0; i < quorum->members.size(); i++) {
auto h = ::SerializeHash(std::make_pair(quorum->members[i]->proTxHash, id));
v.emplace_back(h, quorum->members[i]);
}
std::sort(v.begin(), v.end());
return v[attempt].second;
}
void CSigSharesManager::CollectSigSharesToRequest(std::unordered_map<NodeId, std::unordered_map<uint256, CSigSharesInv, StaticSaltedHasher>>& sigSharesToRequest)
{
AssertLockHeld(cs);
@ -928,6 +1014,43 @@ void CSigSharesManager::CollectSigSharesToSend(std::unordered_map<NodeId, std::u
}
}
void CSigSharesManager::CollectSigSharesToSend(std::unordered_map<NodeId, std::vector<CSigShare>>& sigSharesToSend, const std::vector<CNode*>& vNodes)
{
AssertLockHeld(cs);
std::unordered_map<uint256, CNode*> proTxToNode;
for (CNode* pnode : vNodes) {
if (pnode->verifiedProRegTxHash.IsNull()) {
continue;
}
proTxToNode.emplace(pnode->verifiedProRegTxHash, pnode);
}
auto curTime = GetAdjustedTime();
for (auto& p : signedSessions) {
if (p.second.attempt > p.second.quorum->params.recoveryMembers) {
continue;
}
if (curTime >= p.second.nextAttemptTime) {
p.second.nextAttemptTime = curTime + SEND_FOR_RECOVERY_TIMEOUT;
auto dmn = SelectMemberForRecovery(p.second.quorum, p.second.sigShare.id, p.second.attempt);
LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- sending to %s, signHash=%s\n", __func__,
dmn->proTxHash.ToString(), p.second.sigShare.GetSignHash().ToString());
p.second.attempt++;
auto it = proTxToNode.find(dmn->proTxHash);
if (it == proTxToNode.end()) {
continue;
}
auto& m = sigSharesToSend[it->second->GetId()];
m.emplace_back(p.second.sigShare);
}
}
}
void CSigSharesManager::CollectSigSharesToAnnounce(std::unordered_map<NodeId, std::unordered_map<uint256, CSigSharesInv, StaticSaltedHasher>>& sigSharesToAnnounce)
{
AssertLockHeld(cs);
@ -984,6 +1107,7 @@ bool CSigSharesManager::SendMessages()
{
std::unordered_map<NodeId, std::unordered_map<uint256, CSigSharesInv, StaticSaltedHasher>> sigSharesToRequest;
std::unordered_map<NodeId, std::unordered_map<uint256, CBatchedSigShares, StaticSaltedHasher>> sigShareBatchesToSend;
std::unordered_map<NodeId, std::vector<CSigShare>> sigSharesToSend;
std::unordered_map<NodeId, std::unordered_map<uint256, CSigSharesInv, StaticSaltedHasher>> sigSharesToAnnounce;
std::unordered_map<NodeId, std::vector<CSigSesAnn>> sigSessionAnnouncements;
@ -1006,11 +1130,17 @@ bool CSigSharesManager::SendMessages()
return session->sendSessionId;
};
std::vector<CNode*> vNodesCopy = g_connman->CopyNodeVector(CConnman::FullyConnectedOnly);
{
LOCK(cs);
CollectSigSharesToRequest(sigSharesToRequest);
CollectSigSharesToSend(sigShareBatchesToSend);
CollectSigSharesToAnnounce(sigSharesToAnnounce);
if (!sporkManager.IsSporkActive(SPORK_21_QUORUM_ALL_CONNECTED)) {
CollectSigSharesToRequest(sigSharesToRequest);
CollectSigSharesToSend(sigShareBatchesToSend);
CollectSigSharesToAnnounce(sigSharesToAnnounce);
} else {
CollectSigSharesToSend(sigSharesToSend, vNodesCopy);
}
for (auto& p : sigSharesToRequest) {
for (auto& p2 : p.second) {
@ -1031,8 +1161,6 @@ bool CSigSharesManager::SendMessages()
bool didSend = false;
std::vector<CNode*> vNodesCopy = g_connman->CopyNodeVector(CConnman::FullyConnectedOnly);
for (auto& pnode : vNodesCopy) {
CNetMsgMaker msgMaker(pnode->GetSendVersion());
@ -1119,6 +1247,25 @@ bool CSigSharesManager::SendMessages()
didSend = true;
}
}
auto lt = sigSharesToSend.find(pnode->GetId());
if (lt != sigSharesToSend.end()) {
std::vector<CSigShare> msgs;
for (auto& sigShare : lt->second) {
LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::SendMessages -- QSIGSHARE signHash=%s, node=%d\n",
sigShare.GetSignHash().ToString(), pnode->GetId());
msgs.emplace_back(std::move(sigShare));
if (msgs.size() == MAX_MSGS_SIG_SHARES) {
g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARE, msgs), false);
msgs.clear();
didSend = true;
}
}
if (!msgs.empty()) {
g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARE, msgs), false);
didSend = true;
}
}
}
// looped through all nodes, release them
@ -1285,6 +1432,7 @@ void CSigSharesManager::RemoveSigSharesForSession(const uint256& signHash)
sigSharesRequested.EraseAllForSignHash(signHash);
sigSharesToAnnounce.EraseAllForSignHash(signHash);
sigShares.EraseAllForSignHash(signHash);
signedSessions.erase(signHash);
timeSeenForSessions.erase(signHash);
}
@ -1431,6 +1579,15 @@ void CSigSharesManager::Sign(const CQuorumCPtr& quorum, const uint256& id, const
LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- signed sigShare. signHash=%s, id=%s, msgHash=%s, llmqType=%d, quorum=%s, time=%s\n", __func__,
signHash.ToString(), sigShare.id.ToString(), sigShare.msgHash.ToString(), quorum->params.type, quorum->qc.quorumHash.ToString(), t.count());
ProcessSigShare(-1, sigShare, *g_connman, quorum);
if (sporkManager.IsSporkActive(SPORK_21_QUORUM_ALL_CONNECTED)) {
LOCK(cs);
auto& session = signedSessions[sigShare.GetSignHash()];
session.sigShare = sigShare;
session.quorum = quorum;
session.nextAttemptTime = 0;
session.attempt = 0;
}
}
// causes all known sigShares to be re-announced

View File

@ -30,7 +30,6 @@ namespace llmq
// <signHash, quorumMember>
typedef std::pair<uint256, uint16_t> SigShareKey;
// this one does not get transmitted over the wire as it is batched inside CBatchedSigShares
class CSigShare
{
public:
@ -54,6 +53,22 @@ public:
assert(!key.first.IsNull());
return key.first;
}
ADD_SERIALIZE_METHODS
template<typename Stream, typename Operation>
inline void SerializationOp(Stream& s, Operation ser_action) {
READWRITE(llmqType);
READWRITE(quorumHash);
READWRITE(quorumMember);
READWRITE(id);
READWRITE(msgHash);
READWRITE(sigShare);
if (ser_action.ForRead()) {
UpdateKey();
}
}
};
// Nodes will first announce a signing session with a sessionId to be used in all future P2P messages related to that
@ -327,6 +342,16 @@ public:
void RemoveSession(const uint256& signHash);
};
class CSignedSession
{
public:
CSigShare sigShare;
CQuorumCPtr quorum;
int64_t nextAttemptTime{0};
int attempt{0};
};
class CSigSharesManager : public CRecoveredSigsListener
{
static const int64_t SESSION_NEW_SHARES_TIMEOUT = 60;
@ -339,6 +364,9 @@ class CSigSharesManager : public CRecoveredSigsListener
// 400 is the maximum quorum size, so this is also the maximum number of sigs we need to support
const size_t MAX_MSGS_TOTAL_BATCHED_SIGS = 400;
const int64_t SEND_FOR_RECOVERY_TIMEOUT = 1;
const size_t MAX_MSGS_SIG_SHARES = 32;
private:
CCriticalSection cs;
@ -346,6 +374,7 @@ private:
CThreadInterrupt workInterrupt;
SigShareMap<CSigShare> sigShares;
std::unordered_map<uint256, CSignedSession, StaticSaltedHasher> signedSessions;
// stores time of last receivedSigShare. Used to detect timeouts
std::unordered_map<uint256, int64_t, StaticSaltedHasher> timeSeenForSessions;
@ -381,12 +410,15 @@ public:
void HandleNewRecoveredSig(const CRecoveredSig& recoveredSig);
static CDeterministicMNCPtr SelectMemberForRecovery(const CQuorumCPtr& quorum, const uint256& id, int attempt);
private:
// all of these return false when the currently processed message should be aborted (as each message actually contains multiple messages)
bool ProcessMessageSigSesAnn(CNode* pfrom, const CSigSesAnn& ann, CConnman& connman);
bool ProcessMessageSigSharesInv(CNode* pfrom, const CSigSharesInv& inv, CConnman& connman);
bool ProcessMessageGetSigShares(CNode* pfrom, const CSigSharesInv& inv, CConnman& connman);
bool ProcessMessageBatchedSigShares(CNode* pfrom, const CBatchedSigShares& batchedSigShares, CConnman& connman);
void ProcessMessageSigShare(NodeId fromId, const CSigShare& sigShare, CConnman& connman);
bool VerifySigSharesInv(NodeId from, Consensus::LLMQType llmqType, const CSigSharesInv& inv);
bool PreVerifyBatchedSigShares(NodeId nodeId, const CSigSharesNodeState::SessionInfo& session, const CBatchedSigShares& batchedSigShares, bool& retBan);
@ -417,6 +449,7 @@ private:
bool SendMessages();
void CollectSigSharesToRequest(std::unordered_map<NodeId, std::unordered_map<uint256, CSigSharesInv, StaticSaltedHasher>>& sigSharesToRequest);
void CollectSigSharesToSend(std::unordered_map<NodeId, std::unordered_map<uint256, CBatchedSigShares, StaticSaltedHasher>>& sigSharesToSend);
void CollectSigSharesToSend(std::unordered_map<NodeId, std::vector<CSigShare>>& sigSharesToSend, const std::vector<CNode*>& vNodes);
void CollectSigSharesToAnnounce(std::unordered_map<NodeId, std::unordered_map<uint256, CSigSharesInv, StaticSaltedHasher>>& sigSharesToAnnounce);
bool SignPendingSigShares();
void WorkThreadMain();

View File

@ -69,6 +69,7 @@ const char *QSIGSHARESINV="qsigsinv";
const char *QGETSIGSHARES="qgetsigs";
const char *QBSIGSHARES="qbsigs";
const char *QSIGREC="qsigrec";
const char *QSIGSHARE="qsigshare";
const char *CLSIG="clsig";
const char *ISLOCK="islock";
const char *MNAUTH="mnauth";
@ -135,6 +136,7 @@ const static std::string allNetMessageTypes[] = {
NetMsgType::QGETSIGSHARES,
NetMsgType::QBSIGSHARES,
NetMsgType::QSIGREC,
NetMsgType::QSIGSHARE,
NetMsgType::CLSIG,
NetMsgType::ISLOCK,
NetMsgType::MNAUTH,

View File

@ -266,6 +266,7 @@ extern const char *QSIGSHARESINV;
extern const char *QGETSIGSHARES;
extern const char *QBSIGSHARES;
extern const char *QSIGREC;
extern const char *QSIGSHARE;
extern const char *CLSIG;
extern const char *ISLOCK;
extern const char *MNAUTH;