Fix issue with consensus payload signature and slightly fix pre_prepare consensus code handling

This commit is contained in:
ChronosX88 2020-11-21 01:29:30 +04:00
parent d38bc81dcf
commit b2e956d1d4
Signed by: ChronosXYZ
GPG Key ID: 085A69A82C8C511A
4 changed files with 82 additions and 30 deletions

View File

@ -2,6 +2,9 @@ package consensus
import ( import (
"math/big" "math/big"
"sync"
"github.com/Secured-Finance/dione/sigs"
"github.com/Secured-Finance/dione/consensus/types" "github.com/Secured-Finance/dione/consensus/types"
@ -11,29 +14,29 @@ import (
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
"github.com/Secured-Finance/dione/pubsub" "github.com/Secured-Finance/dione/pubsub"
types2 "github.com/Secured-Finance/dione/types"
) )
type PBFTConsensusManager struct { type PBFTConsensusManager struct {
psb *pubsub.PubSubRouter psb *pubsub.PubSubRouter
//Consensuses map[string]*ConsensusData
//maxFaultNodes int
minApprovals int minApprovals int
privKey []byte privKey []byte
prePreparePool *PrePreparePool prePreparePool *PrePreparePool
preparePool *PreparePool preparePool *PreparePool
commitPool *CommitPool commitPool *CommitPool
consensusLeaders map[string]bool consensusInfo map[string]*ConsensusData
ethereumClient *ethclient.EthereumClient ethereumClient *ethclient.EthereumClient
} }
//type ConsensusData struct { type ConsensusData struct {
// preparedCount int // preparedCount int
// commitCount int // commitCount int
// mutex sync.Mutex mutex sync.Mutex
alreadySubmitted bool
// result string // result string
// test bool // test bool
// onConsensusFinishCallback func(finalData string) // onConsensusFinishCallback func(finalData string)
//} }
func NewPBFTConsensusManager(psb *pubsub.PubSubRouter, minApprovals int, privKey []byte, ethereumClient *ethclient.EthereumClient) *PBFTConsensusManager { func NewPBFTConsensusManager(psb *pubsub.PubSubRouter, minApprovals int, privKey []byte, ethereumClient *ethclient.EthereumClient) *PBFTConsensusManager {
pcm := &PBFTConsensusManager{} pcm := &PBFTConsensusManager{}
@ -44,7 +47,7 @@ func NewPBFTConsensusManager(psb *pubsub.PubSubRouter, minApprovals int, privKey
pcm.minApprovals = minApprovals pcm.minApprovals = minApprovals
pcm.privKey = privKey pcm.privKey = privKey
pcm.ethereumClient = ethereumClient pcm.ethereumClient = ethereumClient
pcm.consensusLeaders = map[string]bool{} pcm.consensusInfo = map[string]*ConsensusData{}
pcm.psb.Hook(types.MessageTypePrePrepare, pcm.handlePrePrepare) pcm.psb.Hook(types.MessageTypePrePrepare, pcm.handlePrePrepare)
pcm.psb.Hook(types.MessageTypePrepare, pcm.handlePrepare) pcm.psb.Hook(types.MessageTypePrepare, pcm.handlePrepare)
pcm.psb.Hook(types.MessageTypeCommit, pcm.handleCommit) pcm.psb.Hook(types.MessageTypeCommit, pcm.handleCommit)
@ -185,7 +188,7 @@ func NewPBFTConsensusManager(psb *pubsub.PubSubRouter, minApprovals int, privKey
//} //}
func (pcm *PBFTConsensusManager) Propose(consensusID, data string, requestID *big.Int, callbackAddress common.Address) error { func (pcm *PBFTConsensusManager) Propose(consensusID, data string, requestID *big.Int, callbackAddress common.Address) error {
pcm.consensusLeaders[consensusID] = true pcm.consensusInfo[consensusID] = &ConsensusData{}
reqIDRaw := requestID.String() reqIDRaw := requestID.String()
callbackAddressHex := callbackAddress.Hex() callbackAddressHex := callbackAddress.Hex()
prePrepareMsg, err := pcm.prePreparePool.CreatePrePrepare(consensusID, data, reqIDRaw, callbackAddressHex, pcm.privKey) prePrepareMsg, err := pcm.prePreparePool.CreatePrePrepare(consensusID, data, reqIDRaw, callbackAddressHex, pcm.privKey)
@ -206,7 +209,19 @@ func (pcm *PBFTConsensusManager) handlePrePrepare(message *types.Message) {
return return
} }
pcm.psb.BroadcastToServiceTopic(message) pcm.prePreparePool.AddPrePrepare(message)
err := pcm.resignMessage(message)
if err != nil {
logrus.Errorf(err.Error())
return
}
err = pcm.psb.BroadcastToServiceTopic(message)
if err != nil {
logrus.Errorf(err.Error())
return
}
prepareMsg, err := pcm.preparePool.CreatePrepare(message, pcm.privKey) prepareMsg, err := pcm.preparePool.CreatePrepare(message, pcm.privKey)
if err != nil { if err != nil {
logrus.Errorf("failed to create prepare message: %w", err) logrus.Errorf("failed to create prepare message: %w", err)
@ -225,7 +240,16 @@ func (pcm *PBFTConsensusManager) handlePrepare(message *types.Message) {
} }
pcm.preparePool.AddPrepare(message) pcm.preparePool.AddPrepare(message)
pcm.psb.BroadcastToServiceTopic(message) err := pcm.resignMessage(message)
if err != nil {
logrus.Errorf(err.Error())
return
}
err = pcm.psb.BroadcastToServiceTopic(message)
if err != nil {
logrus.Errorf(err.Error())
return
}
if pcm.preparePool.PrepareSize(message.Payload.ConsensusID) >= pcm.minApprovals { if pcm.preparePool.PrepareSize(message.Payload.ConsensusID) >= pcm.minApprovals {
commitMsg, err := pcm.commitPool.CreateCommit(message, pcm.privKey) commitMsg, err := pcm.commitPool.CreateCommit(message, pcm.privKey)
@ -247,11 +271,25 @@ func (pcm *PBFTConsensusManager) handleCommit(message *types.Message) {
} }
pcm.commitPool.AddCommit(message) pcm.commitPool.AddCommit(message)
pcm.psb.BroadcastToServiceTopic(message) err := pcm.resignMessage(message)
if err != nil {
logrus.Errorf(err.Error())
return
}
err = pcm.psb.BroadcastToServiceTopic(message)
if err != nil {
logrus.Errorf(err.Error())
return
}
consensusMsg := message.Payload consensusMsg := message.Payload
if pcm.commitPool.CommitSize(consensusMsg.ConsensusID) >= pcm.minApprovals { if pcm.commitPool.CommitSize(consensusMsg.ConsensusID) >= pcm.minApprovals {
if pcm.consensusLeaders[consensusMsg.ConsensusID] { if info, ok := pcm.consensusInfo[consensusMsg.ConsensusID]; ok {
info.mutex.Lock()
defer info.mutex.Unlock()
if info.alreadySubmitted {
return
}
logrus.Infof("Submitting on-chain result for consensus ID: %s", consensusMsg.ConsensusID) logrus.Infof("Submitting on-chain result for consensus ID: %s", consensusMsg.ConsensusID)
reqID, ok := new(big.Int).SetString(consensusMsg.RequestID, 10) reqID, ok := new(big.Int).SetString(consensusMsg.RequestID, 10)
if !ok { if !ok {
@ -262,6 +300,16 @@ func (pcm *PBFTConsensusManager) handleCommit(message *types.Message) {
if err != nil { if err != nil {
logrus.Errorf("Failed to submit on-chain result: %w", err) logrus.Errorf("Failed to submit on-chain result: %w", err)
} }
info.alreadySubmitted = true
} }
} }
} }
func (pcm *PBFTConsensusManager) resignMessage(msg *types.Message) error {
sig, err := sigs.Sign(types2.SigTypeEd25519, pcm.privKey, []byte(msg.Payload.Data))
if err != nil {
return err
}
msg.Payload.Signature = sig.Data
return nil
}

View File

@ -106,13 +106,12 @@ func (m *Miner) MineTask(ctx context.Context, event *oracleEmitter.OracleEmitter
if err != nil { if err != nil {
return nil, xerrors.Errorf("Couldn't get solana request: %w", err) return nil, xerrors.Errorf("Couldn't get solana request: %w", err)
} }
response := res.Body()
var txRes solTypes.TxResponse var txRes solTypes.TxResponse
if err = json.Unmarshal(response, &txRes); err != nil { if err = json.Unmarshal(res, &txRes); err != nil {
return nil, xerrors.Errorf("Couldn't unmarshal solana response: %w", err) return nil, xerrors.Errorf("Couldn't unmarshal solana response: %w", err)
} }
blockHash := txRes.Result.Transaction.Message.RecentBlockhash blockHash := txRes.Result.Transaction.Message.RecentBlockhash
signature, err := sign(ctx, m.address, response) signature, err := sign(ctx, m.address, res)
if err != nil { if err != nil {
return nil, xerrors.Errorf("Couldn't sign solana response: %w", err) return nil, xerrors.Errorf("Couldn't sign solana response: %w", err)
} }
@ -122,7 +121,7 @@ func (m *Miner) MineTask(ctx context.Context, event *oracleEmitter.OracleEmitter
Ticket: ticket, Ticket: ticket,
ElectionProof: winner, ElectionProof: winner,
BeaconEntries: bvals, BeaconEntries: bvals,
Payload: response, Payload: res,
BlockHash: blockHash, BlockHash: blockHash,
Signature: signature, Signature: signature,
DrandRound: types.DrandRound(rbase.Round), DrandRound: types.DrandRound(rbase.Round),

View File

@ -41,7 +41,7 @@ func (n *Node) subscribeOnEthContracts(ctx context.Context) {
// } // }
//}) //})
err = n.ConsensusManager.Propose(task.BlockHash, task.BlockHash, event.RequestID, event.CallbackAddress) err = n.ConsensusManager.Propose(event.RequestID.String(), task.BlockHash, event.RequestID, event.CallbackAddress)
if err != nil { if err != nil {
logrus.Errorf("Failed to propose task: %w", err) logrus.Errorf("Failed to propose task: %w", err)
} }

View File

@ -38,7 +38,12 @@ func (ed25519Signer) Verify(sig []byte, a peer.ID, msg []byte) error {
return err return err
} }
if valid, err := pubKey.Verify(msg, sig); err != nil || !valid { pKeyRaw, err := pubKey.Raw()
if err != nil {
return err
}
if valid := ed25519.Verify(pKeyRaw, msg, sig); !valid {
return xerrors.Errorf("failed to verify signature") return xerrors.Errorf("failed to verify signature")
} }
return nil return nil