Implement handler for NewBlock message in SyncManager, refactor block validation architecture

This commit is contained in:
ChronosX88 2021-07-19 23:19:06 +03:00
parent f59aaa6cf2
commit a6cf757fcf
Signed by: ChronosXYZ
GPG Key ID: 085A69A82C8C511A
14 changed files with 468 additions and 416 deletions

View File

@ -0,0 +1,8 @@
package beacon
// RandomnessType specifies a type of randomness.
type RandomnessType int64
const (
RandomnessTypeElectionProofProduction RandomnessType = 1 + iota
)

29
beacon/utils.go Normal file
View File

@ -0,0 +1,29 @@
package beacon
import (
"encoding/binary"
"github.com/minio/blake2b-simd"
"golang.org/x/xerrors"
)
func DrawRandomness(rbase []byte, randomnessType RandomnessType, round uint64, entropy []byte) ([]byte, error) {
h := blake2b.New256()
if err := binary.Write(h, binary.BigEndian, int64(randomnessType)); err != nil {
return nil, xerrors.Errorf("deriving randomness: %v", err)
}
VRFDigest := blake2b.Sum256(rbase)
_, err := h.Write(VRFDigest[:])
if err != nil {
return nil, xerrors.Errorf("hashing VRFDigest: %w", err)
}
if err := binary.Write(h, binary.BigEndian, round); err != nil {
return nil, xerrors.Errorf("deriving randomness: %v", err)
}
_, err = h.Write(entropy)
if err != nil {
return nil, xerrors.Errorf("hashing entropy: %v", err)
}
return h.Sum(nil), nil
}

28
beacon/vrf.go Normal file
View File

@ -0,0 +1,28 @@
package beacon
import (
"fmt"
"github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/peer"
)
func ComputeVRF(privKey crypto.PrivKey, sigInput []byte) ([]byte, error) {
return privKey.Sign(sigInput)
}
func VerifyVRF(worker peer.ID, vrfBase, vrfproof []byte) error {
pk, err := worker.ExtractPublicKey()
if err != nil {
return err
}
ok, err := pk.Verify(vrfBase, vrfproof)
if err != nil {
return err
}
if !ok {
return fmt.Errorf("vrf was invalid")
}
return nil
}

View File

@ -1,10 +1,22 @@
package blockchain package blockchain
import ( import (
"bytes"
"context"
"encoding/binary" "encoding/binary"
"encoding/hex" "encoding/hex"
"errors" "errors"
"fmt"
"os" "os"
"sync"
"github.com/Secured-Finance/dione/beacon"
"github.com/Secured-Finance/dione/consensus/validation"
"github.com/Secured-Finance/dione/types"
"github.com/sirupsen/logrus"
"github.com/wealdtech/go-merkletree"
"github.com/wealdtech/go-merkletree/keccak256"
"github.com/asaskevich/EventBus" "github.com/asaskevich/EventBus"
@ -29,16 +41,22 @@ var (
) )
type BlockChain struct { type BlockChain struct {
// db-related
dbEnv *lmdb.Env dbEnv *lmdb.Env
db lmdb.DBI db lmdb.DBI
metadataIndex *utils.Index metadataIndex *utils.Index
heightIndex *utils.Index heightIndex *utils.Index
bus EventBus.Bus
bus EventBus.Bus
miner *Miner
b beacon.BeaconAPI
} }
func NewBlockChain(path string, bus EventBus.Bus) (*BlockChain, error) { func NewBlockChain(path string, bus EventBus.Bus, miner *Miner, b beacon.BeaconAPI) (*BlockChain, error) {
chain := &BlockChain{ chain := &BlockChain{
bus: bus, bus: bus,
miner: miner,
b: b,
} }
// configure lmdb env // configure lmdb env
@ -88,16 +106,16 @@ func NewBlockChain(path string, bus EventBus.Bus) (*BlockChain, error) {
return chain, nil return chain, nil
} }
func (bp *BlockChain) setLatestBlockHeight(height uint64) error { func (bc *BlockChain) setLatestBlockHeight(height uint64) error {
err := bp.metadataIndex.PutUint64([]byte(LatestBlockHeightKey), height) err := bc.metadataIndex.PutUint64([]byte(LatestBlockHeightKey), height)
if err != nil { if err != nil {
return err return err
} }
return nil return nil
} }
func (bp *BlockChain) GetLatestBlockHeight() (uint64, error) { func (bc *BlockChain) GetLatestBlockHeight() (uint64, error) {
height, err := bp.metadataIndex.GetUint64([]byte(LatestBlockHeightKey)) height, err := bc.metadataIndex.GetUint64([]byte(LatestBlockHeightKey))
if err != nil { if err != nil {
if err == utils.ErrIndexKeyNotFound { if err == utils.ErrIndexKeyNotFound {
return 0, ErrLatestHeightNil return 0, ErrLatestHeightNil
@ -107,8 +125,20 @@ func (bp *BlockChain) GetLatestBlockHeight() (uint64, error) {
return height, nil return height, nil
} }
func (bp *BlockChain) StoreBlock(block *types2.Block) error { func (bc *BlockChain) StoreBlock(block *types2.Block) error {
err := bp.dbEnv.Update(func(txn *lmdb.Txn) error { if exists, err := bc.HasBlock(block.Header.Hash); err != nil {
return err
} else if exists {
//return fmt.Errorf("block already exists in blockchain")
return nil
}
err := bc.ValidateBlock(block)
if err != nil {
return fmt.Errorf("failed to store block: %w", err)
}
err = bc.dbEnv.Update(func(txn *lmdb.Txn) error {
data, err := cbor.Marshal(block.Data) data, err := cbor.Marshal(block.Data)
if err != nil { if err != nil {
return err return err
@ -118,11 +148,11 @@ func (bp *BlockChain) StoreBlock(block *types2.Block) error {
return err return err
} }
blockHash := hex.EncodeToString(block.Header.Hash) blockHash := hex.EncodeToString(block.Header.Hash)
err = txn.Put(bp.db, []byte(DefaultBlockDataPrefix+blockHash), data, 0) err = txn.Put(bc.db, []byte(DefaultBlockDataPrefix+blockHash), data, 0)
if err != nil { if err != nil {
return err return err
} }
err = txn.Put(bp.db, []byte(DefaultBlockHeaderPrefix+blockHash), headerData, 0) // store header separately for easy fetching err = txn.Put(bc.db, []byte(DefaultBlockHeaderPrefix+blockHash), headerData, 0) // store header separately for easy fetching
return err return err
}) })
if err != nil { if err != nil {
@ -132,36 +162,36 @@ func (bp *BlockChain) StoreBlock(block *types2.Block) error {
// update index "height -> block hash" // update index "height -> block hash"
heightBytes := make([]byte, 8) heightBytes := make([]byte, 8)
binary.LittleEndian.PutUint64(heightBytes, block.Header.Height) binary.LittleEndian.PutUint64(heightBytes, block.Header.Height)
err = bp.heightIndex.PutBytes(heightBytes, block.Header.Hash) err = bc.heightIndex.PutBytes(heightBytes, block.Header.Hash)
if err != nil { if err != nil {
return err return err
} }
// update latest block height // update latest block height
height, err := bp.GetLatestBlockHeight() height, err := bc.GetLatestBlockHeight()
if err != nil && err != ErrLatestHeightNil { if err != nil && err != ErrLatestHeightNil {
return err return err
} }
if err == ErrLatestHeightNil || block.Header.Height > height { if err == ErrLatestHeightNil || block.Header.Height > height {
if err = bp.setLatestBlockHeight(block.Header.Height); err != nil { if err = bc.setLatestBlockHeight(block.Header.Height); err != nil {
return err return err
} }
} else if block.Header.Height > height { } else if block.Header.Height > height {
if err = bp.setLatestBlockHeight(block.Header.Height); err != nil { if err = bc.setLatestBlockHeight(block.Header.Height); err != nil {
return err return err
} }
bp.bus.Publish("blockchain:latestBlockHeightUpdated", block) bc.bus.Publish("blockchain:latestBlockHeightUpdated", block)
} }
bp.bus.Publish("blockchain:blockCommitted", block) bc.bus.Publish("blockchain:blockCommitted", block)
return nil return nil
} }
func (bp *BlockChain) HasBlock(blockHash []byte) (bool, error) { func (bc *BlockChain) HasBlock(blockHash []byte) (bool, error) {
var blockExists bool var blockExists bool
err := bp.dbEnv.View(func(txn *lmdb.Txn) error { err := bc.dbEnv.View(func(txn *lmdb.Txn) error {
h := hex.EncodeToString(blockHash) h := hex.EncodeToString(blockHash)
_, err := txn.Get(bp.db, []byte(DefaultBlockHeaderPrefix+h)) // try to fetch block header _, err := txn.Get(bc.db, []byte(DefaultBlockHeaderPrefix+h)) // try to fetch block header
if err != nil { if err != nil {
if lmdb.IsNotFound(err) { if lmdb.IsNotFound(err) {
blockExists = false blockExists = false
@ -178,11 +208,11 @@ func (bp *BlockChain) HasBlock(blockHash []byte) (bool, error) {
return blockExists, nil return blockExists, nil
} }
func (bp *BlockChain) FetchBlockData(blockHash []byte) ([]*types2.Transaction, error) { func (bc *BlockChain) FetchBlockData(blockHash []byte) ([]*types2.Transaction, error) {
var data []*types2.Transaction var data []*types2.Transaction
err := bp.dbEnv.View(func(txn *lmdb.Txn) error { err := bc.dbEnv.View(func(txn *lmdb.Txn) error {
h := hex.EncodeToString(blockHash) h := hex.EncodeToString(blockHash)
blockData, err := txn.Get(bp.db, []byte(DefaultBlockDataPrefix+h)) blockData, err := txn.Get(bc.db, []byte(DefaultBlockDataPrefix+h))
if err != nil { if err != nil {
if lmdb.IsNotFound(err) { if lmdb.IsNotFound(err) {
return ErrBlockNotFound return ErrBlockNotFound
@ -199,11 +229,11 @@ func (bp *BlockChain) FetchBlockData(blockHash []byte) ([]*types2.Transaction, e
return data, nil return data, nil
} }
func (bp *BlockChain) FetchBlockHeader(blockHash []byte) (*types2.BlockHeader, error) { func (bc *BlockChain) FetchBlockHeader(blockHash []byte) (*types2.BlockHeader, error) {
var blockHeader types2.BlockHeader var blockHeader types2.BlockHeader
err := bp.dbEnv.View(func(txn *lmdb.Txn) error { err := bc.dbEnv.View(func(txn *lmdb.Txn) error {
h := hex.EncodeToString(blockHash) h := hex.EncodeToString(blockHash)
data, err := txn.Get(bp.db, []byte(DefaultBlockHeaderPrefix+h)) data, err := txn.Get(bc.db, []byte(DefaultBlockHeaderPrefix+h))
if err != nil { if err != nil {
if lmdb.IsNotFound(err) { if lmdb.IsNotFound(err) {
return ErrBlockNotFound return ErrBlockNotFound
@ -219,15 +249,15 @@ func (bp *BlockChain) FetchBlockHeader(blockHash []byte) (*types2.BlockHeader, e
return &blockHeader, nil return &blockHeader, nil
} }
func (bp *BlockChain) FetchBlock(blockHash []byte) (*types2.Block, error) { func (bc *BlockChain) FetchBlock(blockHash []byte) (*types2.Block, error) {
var block types2.Block var block types2.Block
header, err := bp.FetchBlockHeader(blockHash) header, err := bc.FetchBlockHeader(blockHash)
if err != nil { if err != nil {
return nil, err return nil, err
} }
block.Header = header block.Header = header
data, err := bp.FetchBlockData(blockHash) data, err := bc.FetchBlockData(blockHash)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -236,34 +266,169 @@ func (bp *BlockChain) FetchBlock(blockHash []byte) (*types2.Block, error) {
return &block, nil return &block, nil
} }
func (bp *BlockChain) FetchBlockByHeight(height uint64) (*types2.Block, error) { func (bc *BlockChain) FetchBlockByHeight(height uint64) (*types2.Block, error) {
var heightBytes = make([]byte, 8) var heightBytes = make([]byte, 8)
binary.LittleEndian.PutUint64(heightBytes, height) binary.LittleEndian.PutUint64(heightBytes, height)
blockHash, err := bp.heightIndex.GetBytes(heightBytes) blockHash, err := bc.heightIndex.GetBytes(heightBytes)
if err != nil { if err != nil {
if err == utils.ErrIndexKeyNotFound { if err == utils.ErrIndexKeyNotFound {
return nil, ErrBlockNotFound return nil, ErrBlockNotFound
} }
} }
block, err := bp.FetchBlock(blockHash) block, err := bc.FetchBlock(blockHash)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return block, nil return block, nil
} }
func (bp *BlockChain) FetchBlockHeaderByHeight(height uint64) (*types2.BlockHeader, error) { func (bc *BlockChain) FetchBlockHeaderByHeight(height uint64) (*types2.BlockHeader, error) {
var heightBytes = make([]byte, 8) var heightBytes = make([]byte, 8)
binary.LittleEndian.PutUint64(heightBytes, height) binary.LittleEndian.PutUint64(heightBytes, height)
blockHash, err := bp.heightIndex.GetBytes(heightBytes) blockHash, err := bc.heightIndex.GetBytes(heightBytes)
if err != nil { if err != nil {
if err == utils.ErrIndexKeyNotFound { if err == utils.ErrIndexKeyNotFound {
return nil, ErrBlockNotFound return nil, ErrBlockNotFound
} }
} }
blockHeader, err := bp.FetchBlockHeader(blockHash) blockHeader, err := bc.FetchBlockHeader(blockHash)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return blockHeader, nil return blockHeader, nil
} }
func (bc *BlockChain) ValidateBlock(block *types2.Block) error {
// === verify block signature ===
pubkey, err := block.Header.Proposer.ExtractPublicKey()
if err != nil {
return fmt.Errorf("unable to extract public key from block proposer's peer id: %w", err)
}
ok, err := pubkey.Verify(block.Header.Hash, block.Header.Signature)
if err != nil {
return fmt.Errorf("failed to verify block signature: %w", err)
}
if !ok {
return fmt.Errorf("signature of block %x is invalid", block.Header.Hash)
}
/////////////////////////////////
// === check last hash merkle proof ===
latestHeight, err := bc.GetLatestBlockHeight()
if err != nil {
return err
}
previousBlockHeader, err := bc.FetchBlockHeaderByHeight(latestHeight)
if err != nil {
return err
}
if !bytes.Equal(block.Header.LastHash, previousBlockHeader.Hash) {
return fmt.Errorf("block header has invalid last block hash (expected: %x, actual %x)", previousBlockHeader.Hash, block.Header.LastHash)
}
verified, err := merkletree.VerifyProofUsing(previousBlockHeader.Hash, true, block.Header.LastHashProof, [][]byte{block.Header.Hash}, keccak256.New())
if err != nil {
return fmt.Errorf("failed to verify last block hash merkle proof: %w", err)
}
if !verified {
return fmt.Errorf("merkle hash of block doesn't contain hash of previous block")
}
/////////////////////////////////
// === verify election proof wincount preliminarily ===
if block.Header.ElectionProof.WinCount < 1 {
return fmt.Errorf("block proposer %s is not a winner", block.Header.Proposer.String())
}
/////////////////////////////////
// === verify miner's eligibility to propose this task ===
err = bc.miner.IsMinerEligibleToProposeBlock(block.Header.ProposerEth)
if err != nil {
return fmt.Errorf("block proposer is not eligible to propose block: %w", err)
}
/////////////////////////////////
// === verify election proof vrf ===
proposerBuf, err := block.Header.Proposer.MarshalBinary()
if err != nil {
return err
}
res, err := bc.b.Entry(context.TODO(), block.Header.ElectionProof.RandomnessRound)
if err != nil {
return err
}
eproofRandomness, err := beacon.DrawRandomness(
res.Data,
beacon.RandomnessTypeElectionProofProduction,
block.Header.Height,
proposerBuf,
)
if err != nil {
return fmt.Errorf("failed to draw ElectionProof randomness: %w", err)
}
err = beacon.VerifyVRF(*block.Header.Proposer, eproofRandomness, block.Header.ElectionProof.VRFProof)
if err != nil {
return fmt.Errorf("failed to verify election proof vrf: %w", err)
}
//////////////////////////////////////
// === compute wincount locally and verify values ===
mStake, nStake, err := bc.miner.GetStakeInfo(block.Header.ProposerEth)
if err != nil {
return fmt.Errorf("failed to get miner stake: %w", err)
}
actualWinCount := block.Header.ElectionProof.ComputeWinCount(mStake, nStake)
if block.Header.ElectionProof.WinCount != actualWinCount {
return fmt.Errorf("locally computed wincount of block is not matching to the received value")
}
//////////////////////////////////////
// === validate block transactions ===
result := make(chan error)
var wg sync.WaitGroup
for _, v := range block.Data {
wg.Add(1)
go func(v *types2.Transaction, c chan error) {
defer wg.Done()
if err := utils.VerifyTx(block.Header, v); err != nil {
c <- fmt.Errorf("failed to verify tx: %w", err)
return
}
var task types.DioneTask
err = cbor.Unmarshal(v.Data, &task)
if err != nil {
c <- fmt.Errorf("failed to unmarshal transaction payload: %w", err)
return
}
if validationFunc := validation.GetValidationMethod(task.OriginChain, task.RequestType); validationFunc != nil {
if err := validationFunc(&task); err != nil {
c <- fmt.Errorf("payload validation has been failed: %w", err)
return
}
} else {
logrus.WithFields(logrus.Fields{
"originChain": task.OriginChain,
"requestType": task.RequestType,
}).Debug("This origin chain/request type doesn't have any payload validation!")
}
}(v, result)
}
go func() {
wg.Wait()
close(result)
}()
for err := range result {
if err != nil {
return err
}
}
/////////////////////////////////
return nil
}

View File

@ -1,4 +1,4 @@
package consensus package blockchain
import ( import (
"errors" "errors"
@ -6,6 +6,9 @@ import (
"math/big" "math/big"
"sync" "sync"
"github.com/Secured-Finance/dione/beacon"
"github.com/Secured-Finance/dione/types"
"github.com/Secured-Finance/dione/blockchain/pool" "github.com/Secured-Finance/dione/blockchain/pool"
"github.com/libp2p/go-libp2p-core/crypto" "github.com/libp2p/go-libp2p-core/crypto"
@ -96,7 +99,7 @@ func (m *Miner) MineBlock(randomness []byte, randomnessRound uint64, lastBlockHe
return nil, fmt.Errorf("failed to update miner stake: %w", err) return nil, fmt.Errorf("failed to update miner stake: %w", err)
} }
winner, err := IsRoundWinner( winner, err := isRoundWinner(
lastBlockHeader.Height+1, lastBlockHeader.Height+1,
m.address, m.address,
randomness, randomness,
@ -110,6 +113,7 @@ func (m *Miner) MineBlock(randomness []byte, randomnessRound uint64, lastBlockHe
} }
if winner == nil { if winner == nil {
logrus.WithField("height", lastBlockHeader.Height+1).Debug("Block is not mined because we are not leader in consensus round")
return nil, nil return nil, nil
} }
@ -136,3 +140,31 @@ func (m *Miner) IsMinerEligibleToProposeBlock(ethAddress common.Address) error {
} }
return nil return nil
} }
func isRoundWinner(round uint64,
worker peer.ID, randomness []byte, randomnessRound uint64, minerStake, networkStake *big.Int, privKey crypto.PrivKey) (*types.ElectionProof, error) {
buf, err := worker.MarshalBinary()
if err != nil {
return nil, fmt.Errorf("failed to marshal address: %w", err)
}
electionRand, err := beacon.DrawRandomness(randomness, beacon.RandomnessTypeElectionProofProduction, round, buf)
if err != nil {
return nil, fmt.Errorf("failed to draw randomness: %w", err)
}
vrfout, err := beacon.ComputeVRF(privKey, electionRand)
if err != nil {
return nil, fmt.Errorf("failed to compute VRF: %w", err)
}
ep := &types.ElectionProof{VRFProof: vrfout, RandomnessRound: randomnessRound}
j := ep.ComputeWinCount(minerStake, networkStake)
ep.WinCount = j
if j < 1 {
return nil, nil
}
return ep, nil
}

View File

@ -43,7 +43,7 @@ func NewMempool(bus EventBus.Bus) (*Mempool, error) {
func (mp *Mempool) StoreTx(tx *types2.Transaction) error { func (mp *Mempool) StoreTx(tx *types2.Transaction) error {
hashStr := hex.EncodeToString(tx.Hash) hashStr := hex.EncodeToString(tx.Hash)
err := mp.cache.StoreWithTTL(DefaultTxPrefix+hashStr, tx, DefaultTxTTL) err := mp.cache.StoreWithTTL(DefaultTxPrefix+hashStr, tx, DefaultTxTTL)
logrus.Infof("Submitted new transaction in mempool with hash %x", tx.Hash) logrus.WithField("txHash", hex.EncodeToString(tx.Hash)).Info("Submitted new transaction in mempool")
mp.bus.Publish("mempool:transactionAdded", tx) mp.bus.Publish("mempool:transactionAdded", tx)
return err return err
} }
@ -56,7 +56,7 @@ func (mp *Mempool) DeleteTx(txHash []byte) error {
return err return err
} }
mp.cache.Delete(DefaultTxPrefix + hashStr) mp.cache.Delete(DefaultTxPrefix + hashStr)
logrus.Debugf("Deleted transaction from mempool %x", txHash) logrus.WithField("txHash", hex.EncodeToString(txHash)).Debugf("Deleted transaction from mempool")
mp.bus.Publish("mempool:transactionRemoved", tx) mp.bus.Publish("mempool:transactionRemoved", tx)
return nil return nil
} }

View File

@ -3,6 +3,7 @@ package sync
import ( import (
"bytes" "bytes"
"context" "context"
"encoding/hex"
"errors" "errors"
"fmt" "fmt"
"strings" "strings"
@ -51,11 +52,11 @@ type syncManager struct {
bus EventBus.Bus bus EventBus.Bus
} }
func NewSyncManager(bus EventBus.Bus, bp *blockchain.BlockChain, mp *pool.Mempool, p2pRPCClient *gorpc.Client, bootstrapPeer peer.ID, psb *pubsub.PubSubRouter) SyncManager { func NewSyncManager(bus EventBus.Bus, bc *blockchain.BlockChain, mp *pool.Mempool, p2pRPCClient *gorpc.Client, bootstrapPeer peer.ID, psb *pubsub.PubSubRouter) SyncManager {
ctx, cancelFunc := context.WithCancel(context.Background()) ctx, cancelFunc := context.WithCancel(context.Background())
sm := &syncManager{ sm := &syncManager{
bus: bus, bus: bus,
blockpool: bp, blockpool: bc,
mempool: mp, mempool: mp,
ctx: ctx, ctx: ctx,
ctxCancelFunc: cancelFunc, ctxCancelFunc: cancelFunc,
@ -66,6 +67,7 @@ func NewSyncManager(bus EventBus.Bus, bp *blockchain.BlockChain, mp *pool.Mempoo
} }
psb.Hook(pubsub.NewTxMessageType, sm.onNewTransaction) psb.Hook(pubsub.NewTxMessageType, sm.onNewTransaction)
psb.Hook(pubsub.NewBlockMessageType, sm.onNewBlock)
go func() { go func() {
if err := sm.initialSync(); err != nil { if err := sm.initialSync(); err != nil {
@ -217,7 +219,7 @@ func (sm *syncManager) processReceivedBlock(block types2.Block) error {
} }
verified, err := merkletree.VerifyProofUsing(previousBlockHeader.Hash, false, block.Header.LastHashProof, [][]byte{block.Header.Hash}, keccak256.New()) verified, err := merkletree.VerifyProofUsing(previousBlockHeader.Hash, false, block.Header.LastHashProof, [][]byte{block.Header.Hash}, keccak256.New())
if err != nil { if err != nil {
return fmt.Errorf("failed to verify last block hash merkle proof: %s", err.Error()) return fmt.Errorf("failed to verify last block hash merkle proof: %w", err)
} }
if !verified { if !verified {
return fmt.Errorf("merkle hash of current block doesn't contain hash of previous block") return fmt.Errorf("merkle hash of current block doesn't contain hash of previous block")
@ -246,13 +248,30 @@ func (sm *syncManager) onNewTransaction(message *pubsub.PubSubMessage) {
return return
} }
// TODO add more checks on tx
if !tx.ValidateHash() { if !tx.ValidateHash() {
logrus.Warn("failed to validate tx hash, rejecting it") logrus.WithField("txHash", hex.EncodeToString(tx.Hash)).Warn("failed to validate transaction hash, rejecting it")
return return
} // TODO add more checks on tx }
err = sm.mempool.StoreTx(&tx) err = sm.mempool.StoreTx(&tx)
if err != nil { if err != nil {
logrus.Warnf("failed to store incoming transaction in mempool: %s", err.Error()) logrus.Warnf("failed to store incoming transaction in mempool: %s", err.Error())
} }
} }
func (sm *syncManager) onNewBlock(message *pubsub.PubSubMessage) {
var block types2.Block
err := cbor.Unmarshal(message.Payload, &block)
if err != nil {
logrus.WithField("err", err.Error()).Error("failed to unmarshal payload of NewBlock message")
return
}
err = sm.blockpool.StoreBlock(&block)
if err != nil {
logrus.WithField("err", err.Error()).Error("failed to store block from NewBlock message")
return
}
}

View File

@ -1,6 +1,7 @@
package types package types
import ( import (
"encoding/binary"
"time" "time"
"github.com/Secured-Finance/dione/types" "github.com/Secured-Finance/dione/types"
@ -48,13 +49,16 @@ func CreateBlock(lastBlockHeader *BlockHeader, txs []*Transaction, minerEth comm
timestamp := time.Now().Unix() timestamp := time.Now().Unix()
// extract hashes from transactions // extract hashes from transactions
var txHashes [][]byte var merkleHashes [][]byte
for _, tx := range txs { for _, tx := range txs {
txHashes = append(txHashes, tx.Hash) merkleHashes = append(merkleHashes, tx.Hash)
} }
txHashes = append(txHashes, lastBlockHeader.Hash) merkleHashes = append(merkleHashes, lastBlockHeader.Hash)
timestampBytes := make([]byte, 8)
binary.LittleEndian.PutUint64(timestampBytes, uint64(timestamp))
merkleHashes = append(merkleHashes, timestampBytes)
tree, err := merkletree.NewUsing(txHashes, keccak256.New(), false) tree, err := merkletree.NewUsing(merkleHashes, keccak256.New(), true)
if err != nil { if err != nil {
return nil, err return nil, err
} }

View File

@ -12,7 +12,7 @@ func VerifyTx(blockHeader *types.BlockHeader, tx *types.Transaction) error {
if tx.MerkleProof == nil { if tx.MerkleProof == nil {
return fmt.Errorf("block transaction doesn't have merkle proof") return fmt.Errorf("block transaction doesn't have merkle proof")
} }
txProofVerified, err := merkletree.VerifyProofUsing(tx.Hash, false, tx.MerkleProof, [][]byte{blockHeader.Hash}, keccak256.New()) txProofVerified, err := merkletree.VerifyProofUsing(tx.Hash, true, tx.MerkleProof, [][]byte{blockHeader.Hash}, keccak256.New())
if err != nil { if err != nil {
return fmt.Errorf("failed to verify tx hash merkle proof: %s", err.Error()) return fmt.Errorf("failed to verify tx hash merkle proof: %s", err.Error())
} }

View File

@ -6,6 +6,8 @@ import (
"math/big" "math/big"
"sync" "sync"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/Secured-Finance/dione/beacon" "github.com/Secured-Finance/dione/beacon"
"github.com/fxamacker/cbor/v2" "github.com/fxamacker/cbor/v2"
@ -53,11 +55,12 @@ type PBFTConsensusManager struct {
msgLog *ConsensusMessageLog msgLog *ConsensusMessageLog
validator *ConsensusValidator validator *ConsensusValidator
ethereumClient *ethclient.EthereumClient ethereumClient *ethclient.EthereumClient
miner *Miner miner *blockchain.Miner
blockPool *pool.BlockPool blockPool *pool.BlockPool
mempool *pool.Mempool mempool *pool.Mempool
blockchain *blockchain.BlockChain blockchain *blockchain.BlockChain
state *State state *State
address peer.ID
} }
type State struct { type State struct {
@ -75,28 +78,32 @@ func NewPBFTConsensusManager(
minApprovals int, minApprovals int,
privKey crypto.PrivKey, privKey crypto.PrivKey,
ethereumClient *ethclient.EthereumClient, ethereumClient *ethclient.EthereumClient,
miner *Miner, miner *blockchain.Miner,
bc *blockchain.BlockChain, bc *blockchain.BlockChain,
bp *pool.BlockPool, bp *pool.BlockPool,
b beacon.BeaconNetwork, b beacon.BeaconNetwork,
mempool *pool.Mempool, mempool *pool.Mempool,
address peer.ID,
) *PBFTConsensusManager { ) *PBFTConsensusManager {
pcm := &PBFTConsensusManager{} pcm := &PBFTConsensusManager{
pcm.psb = psb psb: psb,
pcm.miner = miner miner: miner,
pcm.validator = NewConsensusValidator(miner, bc, b) validator: NewConsensusValidator(miner, bc, b),
pcm.msgLog = NewConsensusMessageLog() msgLog: NewConsensusMessageLog(),
pcm.minApprovals = minApprovals minApprovals: minApprovals,
pcm.privKey = privKey privKey: privKey,
pcm.ethereumClient = ethereumClient ethereumClient: ethereumClient,
pcm.state = &State{ state: &State{
ready: false, ready: false,
status: StateStatusUnknown, status: StateStatusUnknown,
},
bus: bus,
blockPool: bp,
mempool: mempool,
blockchain: bc,
address: address,
} }
pcm.bus = bus
pcm.blockPool = bp
pcm.mempool = mempool
pcm.blockchain = bc
pcm.psb.Hook(pubsub.PrePrepareMessageType, pcm.handlePrePrepare) pcm.psb.Hook(pubsub.PrePrepareMessageType, pcm.handlePrePrepare)
pcm.psb.Hook(pubsub.PrepareMessageType, pcm.handlePrepare) pcm.psb.Hook(pubsub.PrepareMessageType, pcm.handlePrepare)
pcm.psb.Hook(pubsub.CommitMessageType, pcm.handleCommit) pcm.psb.Hook(pubsub.CommitMessageType, pcm.handleCommit)
@ -120,8 +127,8 @@ func (pcm *PBFTConsensusManager) propose(blk *types3.Block) error {
} }
func (pcm *PBFTConsensusManager) handlePrePrepare(message *pubsub.PubSubMessage) { func (pcm *PBFTConsensusManager) handlePrePrepare(message *pubsub.PubSubMessage) {
pcm.state.mutex.Lock() //pcm.state.mutex.Lock()
defer pcm.state.mutex.Unlock() //defer pcm.state.mutex.Unlock()
var prePrepare types.PrePrepareMessage var prePrepare types.PrePrepareMessage
err := cbor.Unmarshal(message.Payload, &prePrepare) err := cbor.Unmarshal(message.Payload, &prePrepare)
if err != nil { if err != nil {
@ -129,7 +136,7 @@ func (pcm *PBFTConsensusManager) handlePrePrepare(message *pubsub.PubSubMessage)
return return
} }
if *prePrepare.Block.Header.Proposer == pcm.miner.address { if *prePrepare.Block.Header.Proposer == pcm.address {
return return
} }
@ -144,7 +151,7 @@ func (pcm *PBFTConsensusManager) handlePrePrepare(message *pubsub.PubSubMessage)
logrus.Tracef("received existing pre_prepare msg for block %x", cmsg.Block.Header.Hash) logrus.Tracef("received existing pre_prepare msg for block %x", cmsg.Block.Header.Hash)
return return
} }
if !pcm.validator.Valid(cmsg, map[string]interface{}{"randomness": pcm.state.randomness}) { if !pcm.validator.Valid(cmsg) {
logrus.Warnf("received invalid pre_prepare msg for block %x", cmsg.Block.Header.Hash) logrus.Warnf("received invalid pre_prepare msg for block %x", cmsg.Block.Header.Hash)
return return
} }
@ -163,8 +170,8 @@ func (pcm *PBFTConsensusManager) handlePrePrepare(message *pubsub.PubSubMessage)
} }
func (pcm *PBFTConsensusManager) handlePrepare(message *pubsub.PubSubMessage) { func (pcm *PBFTConsensusManager) handlePrepare(message *pubsub.PubSubMessage) {
pcm.state.mutex.Lock() //pcm.state.mutex.Lock()
defer pcm.state.mutex.Unlock() //defer pcm.state.mutex.Unlock()
var prepare types.PrepareMessage var prepare types.PrepareMessage
err := cbor.Unmarshal(message.Payload, &prepare) err := cbor.Unmarshal(message.Payload, &prepare)
if err != nil { if err != nil {
@ -189,7 +196,7 @@ func (pcm *PBFTConsensusManager) handlePrepare(message *pubsub.PubSubMessage) {
return return
} }
if !pcm.validator.Valid(cmsg, nil) { if !pcm.validator.Valid(cmsg) {
logrus.Warnf("received invalid prepare msg for block %x", cmsg.Blockhash) logrus.Warnf("received invalid prepare msg for block %x", cmsg.Blockhash)
return return
} }
@ -208,8 +215,8 @@ func (pcm *PBFTConsensusManager) handlePrepare(message *pubsub.PubSubMessage) {
} }
func (pcm *PBFTConsensusManager) handleCommit(message *pubsub.PubSubMessage) { func (pcm *PBFTConsensusManager) handleCommit(message *pubsub.PubSubMessage) {
pcm.state.mutex.Lock() //pcm.state.mutex.Lock()
defer pcm.state.mutex.Unlock() //defer pcm.state.mutex.Unlock()
var commit types.CommitMessage var commit types.CommitMessage
err := cbor.Unmarshal(message.Payload, &commit) err := cbor.Unmarshal(message.Payload, &commit)
if err != nil { if err != nil {
@ -233,7 +240,7 @@ func (pcm *PBFTConsensusManager) handleCommit(message *pubsub.PubSubMessage) {
logrus.Tracef("received existing commit msg for block %x", cmsg.Blockhash) logrus.Tracef("received existing commit msg for block %x", cmsg.Blockhash)
return return
} }
if !pcm.validator.Valid(cmsg, nil) { if !pcm.validator.Valid(cmsg) {
logrus.Warnf("received invalid commit msg for block %x", cmsg.Blockhash) logrus.Warnf("received invalid commit msg for block %x", cmsg.Blockhash)
return return
} }
@ -281,39 +288,8 @@ func (pcm *PBFTConsensusManager) onNewBeaconEntry(entry types2.BeaconEntry) {
// if we are miner of this block // if we are miner of this block
// then post dione tasks to target chains (currently, only Ethereum) // then post dione tasks to target chains (currently, only Ethereum)
if block.Header.Proposer.String() == pcm.miner.address.String() { if block.Header.Proposer.String() == pcm.address.String() {
for _, tx := range block.Data { pcm.submitTasksFromBlock(block)
var task types2.DioneTask
err := cbor.Unmarshal(tx.Data, &task)
if err != nil {
logrus.WithFields(logrus.Fields{
"err": err.Error(),
"txHash": hex.EncodeToString(tx.Hash),
}).Error("Failed to unmarshal transaction payload")
continue // FIXME
}
reqIDNumber, ok := big.NewInt(0).SetString(task.RequestID, 10)
if !ok {
logrus.WithFields(logrus.Fields{
"txHash": hex.EncodeToString(tx.Hash),
}).Error("Failed to parse request id number in Dione task")
continue // FIXME
}
err = pcm.ethereumClient.SubmitRequestAnswer(reqIDNumber, task.Payload)
if err != nil {
logrus.WithFields(logrus.Fields{
"err": err.Error(),
"txHash": hex.EncodeToString(tx.Hash),
"reqID": reqIDNumber.String(),
}).Error("Failed to submit task to ETH chain")
continue // FIXME
}
logrus.WithFields(logrus.Fields{
"txHash": hex.EncodeToString(tx.Hash),
"reqID": reqIDNumber.String(),
}).Debug("Dione task has been sucessfully submitted to ETH chain (DioneOracle contract)")
}
} }
pcm.state.blockHeight = pcm.state.blockHeight + 1 pcm.state.blockHeight = pcm.state.blockHeight + 1
@ -336,7 +312,7 @@ func (pcm *PBFTConsensusManager) onNewBeaconEntry(entry types2.BeaconEntry) {
minedBlock, err := pcm.miner.MineBlock(entry.Data, entry.Round, blockHeader) minedBlock, err := pcm.miner.MineBlock(entry.Data, entry.Round, blockHeader)
if err != nil { if err != nil {
if errors.Is(err, ErrNoTxForBlock) { if errors.Is(err, blockchain.ErrNoTxForBlock) {
logrus.Info("Sealing skipped, no transactions in mempool") logrus.Info("Sealing skipped, no transactions in mempool")
} else { } else {
logrus.Errorf("Failed to mine the block: %s", err.Error()) logrus.Errorf("Failed to mine the block: %s", err.Error())
@ -355,6 +331,41 @@ func (pcm *PBFTConsensusManager) onNewBeaconEntry(entry types2.BeaconEntry) {
} }
} }
func (pcm *PBFTConsensusManager) submitTasksFromBlock(block *types3.Block) {
for _, tx := range block.Data {
var task types2.DioneTask
err := cbor.Unmarshal(tx.Data, &task)
if err != nil {
logrus.WithFields(logrus.Fields{
"err": err.Error(),
"txHash": hex.EncodeToString(tx.Hash),
}).Error("Failed to unmarshal transaction payload")
continue // FIXME
}
reqIDNumber, ok := big.NewInt(0).SetString(task.RequestID, 10)
if !ok {
logrus.WithFields(logrus.Fields{
"txHash": hex.EncodeToString(tx.Hash),
}).Error("Failed to parse request id number in Dione task")
continue // FIXME
}
err = pcm.ethereumClient.SubmitRequestAnswer(reqIDNumber, task.Payload)
if err != nil {
logrus.WithFields(logrus.Fields{
"err": err.Error(),
"txHash": hex.EncodeToString(tx.Hash),
"reqID": reqIDNumber.String(),
}).Error("Failed to submit task to ETH chain")
continue // FIXME
}
logrus.WithFields(logrus.Fields{
"txHash": hex.EncodeToString(tx.Hash),
"reqID": reqIDNumber.String(),
}).Debug("Dione task has been sucessfully submitted to ETH chain (DioneOracle contract)")
}
}
func (pcm *PBFTConsensusManager) commitAcceptedBlocks() (*types3.Block, error) { func (pcm *PBFTConsensusManager) commitAcceptedBlocks() (*types3.Block, error) {
blocks := pcm.blockPool.GetAllAcceptedBlocks() blocks := pcm.blockPool.GetAllAcceptedBlocks()
if blocks == nil { if blocks == nil {

View File

@ -1,223 +1,61 @@
package consensus package consensus
import ( import (
"bytes" "encoding/hex"
"context"
"fmt"
"sync"
"github.com/Secured-Finance/dione/beacon" "github.com/Secured-Finance/dione/beacon"
"github.com/sirupsen/logrus"
types3 "github.com/Secured-Finance/dione/blockchain/types"
"github.com/Secured-Finance/dione/blockchain" "github.com/Secured-Finance/dione/blockchain"
"github.com/Secured-Finance/dione/blockchain/utils"
types2 "github.com/Secured-Finance/dione/consensus/types" types2 "github.com/Secured-Finance/dione/consensus/types"
"github.com/Secured-Finance/dione/consensus/validation"
"github.com/Secured-Finance/dione/types"
"github.com/filecoin-project/go-state-types/crypto"
"github.com/fxamacker/cbor/v2"
"github.com/sirupsen/logrus"
"github.com/wealdtech/go-merkletree"
"github.com/wealdtech/go-merkletree/keccak256"
) )
type ConsensusValidator struct { type ConsensusValidator struct {
validationFuncMap map[types2.ConsensusMessageType]func(msg types2.ConsensusMessage, metadata map[string]interface{}) bool validationFuncMap map[types2.ConsensusMessageType]func(msg types2.ConsensusMessage) bool
miner *Miner miner *blockchain.Miner
beacon beacon.BeaconNetwork beacon beacon.BeaconNetwork
blockchain *blockchain.BlockChain blockchain *blockchain.BlockChain
} }
func NewConsensusValidator(miner *Miner, bc *blockchain.BlockChain, b beacon.BeaconNetwork) *ConsensusValidator { func NewConsensusValidator(miner *blockchain.Miner, bc *blockchain.BlockChain, b beacon.BeaconNetwork) *ConsensusValidator {
cv := &ConsensusValidator{ cv := &ConsensusValidator{
miner: miner, miner: miner,
blockchain: bc, blockchain: bc,
beacon: b, beacon: b,
} }
cv.validationFuncMap = map[types2.ConsensusMessageType]func(msg types2.ConsensusMessage, metadata map[string]interface{}) bool{ cv.validationFuncMap = map[types2.ConsensusMessageType]func(msg types2.ConsensusMessage) bool{
// FIXME it all types2.ConsensusMessageTypePrePrepare: func(msg types2.ConsensusMessage) bool {
types2.ConsensusMessageTypePrePrepare: func(msg types2.ConsensusMessage, metadata map[string]interface{}) bool { if err := cv.blockchain.ValidateBlock(msg.Block); err != nil {
// === verify block signature === logrus.WithFields(logrus.Fields{
pubkey, err := msg.Block.Header.Proposer.ExtractPublicKey() "blockHash": hex.EncodeToString(msg.Block.Header.Hash),
if err != nil { "err": err.Error(),
logrus.Errorf("unable to extract public key from block proposer's peer id: %s", err.Error()) }).Error("failed to validate block from PrePrepare message")
return false return false
} }
ok, err := pubkey.Verify(msg.Block.Header.Hash, msg.Block.Header.Signature)
if err != nil {
logrus.Errorf("failed to verify block signature: %s", err.Error())
return false
}
if !ok {
logrus.Errorf("signature of block %x is invalid", msg.Block.Header.Hash)
return false
}
/////////////////////////////////
// === check last hash merkle proof ===
latestHeight, err := cv.blockchain.GetLatestBlockHeight()
if err != nil {
logrus.Error(err)
return false
}
previousBlockHeader, err := cv.blockchain.FetchBlockHeaderByHeight(latestHeight)
if err != nil {
logrus.Error(err)
return false
}
if bytes.Compare(msg.Block.Header.LastHash, previousBlockHeader.Hash) != 0 {
logrus.Errorf("block header has invalid last block hash (expected: %x, actual %x)", previousBlockHeader.Hash, msg.Block.Header.LastHash)
return false
}
verified, err := merkletree.VerifyProofUsing(previousBlockHeader.Hash, false, msg.Block.Header.LastHashProof, [][]byte{msg.Block.Header.Hash}, keccak256.New())
if err != nil {
logrus.Error("failed to verify last block hash merkle proof: %s", err.Error())
return false
}
if !verified {
logrus.Error("merkle hash of current block doesn't contain hash of previous block: %s", err.Error())
return false
}
/////////////////////////////////
// === verify election proof wincount preliminarily ===
if msg.Block.Header.ElectionProof.WinCount < 1 {
logrus.Error("miner isn't a winner!")
return false
}
/////////////////////////////////
// === verify miner's eligibility to propose this task ===
err = cv.miner.IsMinerEligibleToProposeBlock(msg.Block.Header.ProposerEth)
if err != nil {
logrus.Errorf("miner is not eligible to propose block: %v", err)
return false
}
/////////////////////////////////
// === verify election proof vrf ===
proposerBuf, err := msg.Block.Header.Proposer.MarshalBinary()
if err != nil {
logrus.Error(err)
return false
}
res, err := b.Beacon.Entry(context.TODO(), msg.Block.Header.ElectionProof.RandomnessRound)
if err != nil {
logrus.Error(err)
return false
}
eproofRandomness, err := DrawRandomness(
res.Data,
crypto.DomainSeparationTag_ElectionProofProduction,
msg.Block.Header.Height,
proposerBuf,
)
if err != nil {
logrus.Errorf("failed to draw ElectionProof randomness: %s", err.Error())
return false
}
err = VerifyVRF(*msg.Block.Header.Proposer, eproofRandomness, msg.Block.Header.ElectionProof.VRFProof)
if err != nil {
logrus.Errorf("failed to verify election proof vrf: %v", err)
return false
}
//////////////////////////////////////
// === compute wincount locally and verify values ===
mStake, nStake, err := cv.miner.GetStakeInfo(msg.Block.Header.ProposerEth)
if err != nil {
logrus.Errorf("failed to get miner stake: %v", err)
return false
}
actualWinCount := msg.Block.Header.ElectionProof.ComputeWinCount(mStake, nStake)
if msg.Block.Header.ElectionProof.WinCount != actualWinCount {
logrus.Errorf("locally computed wincount of block %x isn't matching received value!", msg.Block.Header.Hash)
return false
}
//////////////////////////////////////
// === validate block transactions ===
result := make(chan error)
var wg sync.WaitGroup
for _, v := range msg.Block.Data {
wg.Add(1)
go func(v *types3.Transaction, c chan error) {
defer wg.Done()
if err := utils.VerifyTx(msg.Block.Header, v); err != nil {
c <- fmt.Errorf("failed to verify tx: %w", err)
return
}
var task types.DioneTask
err = cbor.Unmarshal(v.Data, &task)
if err != nil {
c <- fmt.Errorf("failed to unmarshal transaction payload: %w", err)
return
}
if validationFunc := validation.GetValidationMethod(task.OriginChain, task.RequestType); validationFunc != nil {
if err := validationFunc(&task); err != nil {
c <- fmt.Errorf("payload validation has been failed: %w", err)
return
}
} else {
logrus.WithFields(logrus.Fields{
"originChain": task.OriginChain,
"requestType": task.RequestType,
}).Debug("This origin chain/request type doesn't have any payload validation!")
}
}(v, result)
}
go func() {
wg.Wait()
close(result)
}()
for err := range result {
if err != nil {
logrus.Error(err)
return false
}
}
/////////////////////////////////
return true return true
}, },
types2.ConsensusMessageTypePrepare: func(msg types2.ConsensusMessage, metadata map[string]interface{}) bool { types2.ConsensusMessageTypePrepare: checkSignatureForBlockhash,
pubKey, err := msg.From.ExtractPublicKey() types2.ConsensusMessageTypeCommit: checkSignatureForBlockhash,
if err != nil {
// TODO logging
return false
}
ok, err := pubKey.Verify(msg.Blockhash, msg.Signature)
if err != nil {
// TODO logging
return false
}
return ok
},
types2.ConsensusMessageTypeCommit: func(msg types2.ConsensusMessage, metadata map[string]interface{}) bool {
pubKey, err := msg.From.ExtractPublicKey()
if err != nil {
// TODO logging
return false
}
ok, err := pubKey.Verify(msg.Blockhash, msg.Signature)
if err != nil {
// TODO logging
return false
}
return ok
},
} }
return cv return cv
} }
func (cv *ConsensusValidator) Valid(msg types2.ConsensusMessage, metadata map[string]interface{}) bool { func (cv *ConsensusValidator) Valid(msg types2.ConsensusMessage) bool {
return cv.validationFuncMap[msg.Type](msg, metadata) return cv.validationFuncMap[msg.Type](msg)
}
func checkSignatureForBlockhash(msg types2.ConsensusMessage) bool {
pubKey, err := msg.From.ExtractPublicKey()
if err != nil {
// TODO logging
return false
}
ok, err := pubKey.Verify(msg.Blockhash, msg.Signature)
if err != nil {
// TODO logging
return false
}
return ok
} }

View File

@ -1,9 +1,7 @@
package consensus package consensus
import ( import (
"encoding/binary"
"fmt" "fmt"
"math/big"
"github.com/fxamacker/cbor/v2" "github.com/fxamacker/cbor/v2"
@ -11,87 +9,9 @@ import (
types2 "github.com/Secured-Finance/dione/consensus/types" types2 "github.com/Secured-Finance/dione/consensus/types"
"github.com/minio/blake2b-simd"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/Secured-Finance/dione/types"
crypto2 "github.com/filecoin-project/go-state-types/crypto"
"github.com/libp2p/go-libp2p-core/crypto" "github.com/libp2p/go-libp2p-core/crypto"
"golang.org/x/xerrors"
) )
type SignFunc func(peer.ID, []byte) (*types.Signature, error)
func ComputeVRF(privKey crypto.PrivKey, sigInput []byte) ([]byte, error) {
return privKey.Sign(sigInput)
}
func VerifyVRF(worker peer.ID, vrfBase, vrfproof []byte) error {
pk, err := worker.ExtractPublicKey()
if err != nil {
return err
}
ok, err := pk.Verify(vrfBase, vrfproof)
if err != nil {
return err
}
if !ok {
return fmt.Errorf("vrf was invalid")
}
return nil
}
func IsRoundWinner(round uint64,
worker peer.ID, randomness []byte, randomnessRound uint64, minerStake, networkStake *big.Int, privKey crypto.PrivKey) (*types.ElectionProof, error) {
buf, err := worker.MarshalBinary()
if err != nil {
return nil, fmt.Errorf("failed to marshal address: %w", err)
}
electionRand, err := DrawRandomness(randomness, crypto2.DomainSeparationTag_ElectionProofProduction, round, buf)
if err != nil {
return nil, fmt.Errorf("failed to draw randomness: %w", err)
}
vrfout, err := ComputeVRF(privKey, electionRand)
if err != nil {
return nil, fmt.Errorf("failed to compute VRF: %w", err)
}
ep := &types.ElectionProof{VRFProof: vrfout, RandomnessRound: randomnessRound}
j := ep.ComputeWinCount(minerStake, networkStake)
ep.WinCount = j
if j < 1 {
return nil, nil
}
return ep, nil
}
func DrawRandomness(rbase []byte, pers crypto2.DomainSeparationTag, round uint64, entropy []byte) ([]byte, error) {
h := blake2b.New256()
if err := binary.Write(h, binary.BigEndian, int64(pers)); err != nil {
return nil, xerrors.Errorf("deriving randomness: %v", err)
}
VRFDigest := blake2b.Sum256(rbase)
_, err := h.Write(VRFDigest[:])
if err != nil {
return nil, xerrors.Errorf("hashing VRFDigest: %w", err)
}
if err := binary.Write(h, binary.BigEndian, round); err != nil {
return nil, xerrors.Errorf("deriving randomness: %v", err)
}
_, err = h.Write(entropy)
if err != nil {
return nil, xerrors.Errorf("hashing entropy: %v", err)
}
return h.Sum(nil), nil
}
func NewMessage(cmsg types2.ConsensusMessage, typ types2.ConsensusMessageType, privKey crypto.PrivKey) (*pubsub.PubSubMessage, error) { func NewMessage(cmsg types2.ConsensusMessage, typ types2.ConsensusMessageType, privKey crypto.PrivKey) (*pubsub.PubSubMessage, error) {
var message pubsub.PubSubMessage var message pubsub.PubSubMessage
switch typ { switch typ {

View File

@ -65,7 +65,7 @@ type Node struct {
Config *config.Config Config *config.Config
Ethereum *ethclient.EthereumClient Ethereum *ethclient.EthereumClient
ConsensusManager *consensus.PBFTConsensusManager ConsensusManager *consensus.PBFTConsensusManager
Miner *consensus.Miner Miner *blockchain.Miner
Beacon beacon.BeaconNetwork Beacon beacon.BeaconNetwork
DisputeManager *consensus.DisputeManager DisputeManager *consensus.DisputeManager
BlockPool *pool.BlockPool BlockPool *pool.BlockPool
@ -104,9 +104,10 @@ func NewNode(config *config.Config, prvKey crypto.PrivKey, pexDiscoveryUpdateTim
// initialize ethereum client // initialize ethereum client
ethClient, err := provideEthereumClient(n.Config) ethClient, err := provideEthereumClient(n.Config)
if err != nil { if err != nil {
logrus.Fatal(err) logrus.WithField("err", err.Error()).Fatal("Failed to initialize Ethereum client")
} }
n.Ethereum = ethClient n.Ethereum = ethClient
//goland:noinspection ALL
logrus.WithField("ethAddress", ethClient.GetEthAddress().Hex()).Info("Ethereum client has been initialized!") logrus.WithField("ethAddress", ethClient.GetEthAddress().Hex()).Info("Ethereum client has been initialized!")
// initialize blockchain rpc clients // initialize blockchain rpc clients
@ -135,21 +136,16 @@ func NewNode(config *config.Config, prvKey crypto.PrivKey, pexDiscoveryUpdateTim
n.PeerDiscovery = peerDiscovery n.PeerDiscovery = peerDiscovery
logrus.Info("Peer discovery subsystem has been initialized!") logrus.Info("Peer discovery subsystem has been initialized!")
// initialize event log cache subsystem // initialize random beacon network subsystem
//c := provideCache(config) randomBeaconNetwork, err := provideBeacon(psb.Pubsub, bus)
//n.Cache = c if err != nil {
//logrus.Info("Event cache subsystem has initialized!") logrus.Fatal(err)
}
n.Beacon = randomBeaconNetwork
logrus.Info("Random beacon subsystem has been initialized!")
// == initialize blockchain modules // == initialize blockchain modules
// initialize blockpool database
bc, err := provideBlockChain(n.Config, bus)
if err != nil {
logrus.Fatalf("Failed to initialize blockpool: %s", err.Error())
}
n.BlockChain = bc
logrus.Info("Block pool database has been successfully initialized!")
// initialize mempool // initialize mempool
mp, err := provideMemPool(bus) mp, err := provideMemPool(bus)
if err != nil { if err != nil {
@ -158,6 +154,19 @@ func NewNode(config *config.Config, prvKey crypto.PrivKey, pexDiscoveryUpdateTim
n.MemPool = mp n.MemPool = mp
logrus.Info("Mempool has been successfully initialized!") logrus.Info("Mempool has been successfully initialized!")
// initialize mining subsystem
miner := provideMiner(n.Host.ID(), *n.Ethereum.GetEthAddress(), n.Ethereum, prvKey, mp)
n.Miner = miner
logrus.Info("Mining subsystem has been initialized!")
// initialize blockpool database
bc, err := provideBlockChain(n.Config, bus, miner, randomBeaconNetwork.Beacon)
if err != nil {
logrus.Fatalf("Failed to initialize blockpool: %s", err.Error())
}
n.BlockChain = bc
logrus.Info("Block pool database has been successfully initialized!")
bp, err := provideBlockPool(mp, bus) bp, err := provideBlockPool(mp, bus)
if err != nil { if err != nil {
logrus.Fatalf("Failed to initialize blockpool: %s", err.Error()) logrus.Fatalf("Failed to initialize blockpool: %s", err.Error())
@ -192,21 +201,8 @@ func NewNode(config *config.Config, prvKey crypto.PrivKey, pexDiscoveryUpdateTim
n.SyncManager = sm n.SyncManager = sm
logrus.Info("Blockchain sync subsystem has been successfully initialized!") logrus.Info("Blockchain sync subsystem has been successfully initialized!")
// initialize mining subsystem
miner := provideMiner(n.Host.ID(), *n.Ethereum.GetEthAddress(), n.Ethereum, prvKey, mp)
n.Miner = miner
logrus.Info("Mining subsystem has been initialized!")
// initialize random beacon network subsystem
randomBeaconNetwork, err := provideBeacon(psb.Pubsub, bus)
if err != nil {
logrus.Fatal(err)
}
n.Beacon = randomBeaconNetwork
logrus.Info("Random beacon subsystem has been initialized!")
// initialize consensus subsystem // initialize consensus subsystem
consensusManager := provideConsensusManager(bus, psb, miner, bc, ethClient, prvKey, n.Config.ConsensusMinApprovals, bp, randomBeaconNetwork, mp) consensusManager := provideConsensusManager(bus, psb, miner, bc, ethClient, prvKey, n.Config.ConsensusMinApprovals, bp, randomBeaconNetwork, mp, n.Host.ID())
n.ConsensusManager = consensusManager n.ConsensusManager = consensusManager
logrus.Info("Consensus subsystem has been initialized!") logrus.Info("Consensus subsystem has been initialized!")

View File

@ -60,8 +60,8 @@ func provideDisputeManager(ctx context.Context, ethClient *ethclient.EthereumCli
return consensus.NewDisputeManager(ctx, ethClient, pcm, cfg.Ethereum.DisputeVoteWindow, bc) return consensus.NewDisputeManager(ctx, ethClient, pcm, cfg.Ethereum.DisputeVoteWindow, bc)
} }
func provideMiner(peerID peer.ID, ethAddress common.Address, ethClient *ethclient.EthereumClient, privateKey crypto.PrivKey, mempool *pool.Mempool) *consensus.Miner { func provideMiner(peerID peer.ID, ethAddress common.Address, ethClient *ethclient.EthereumClient, privateKey crypto.PrivKey, mempool *pool.Mempool) *blockchain.Miner {
return consensus.NewMiner(peerID, ethAddress, ethClient, privateKey, mempool) return blockchain.NewMiner(peerID, ethAddress, ethClient, privateKey, mempool)
} }
func provideBeacon(ps *pubsub2.PubSub, bus EventBus.Bus) (beacon.BeaconNetwork, error) { func provideBeacon(ps *pubsub2.PubSub, bus EventBus.Bus) (beacon.BeaconNetwork, error) {
@ -94,7 +94,7 @@ func provideEthereumClient(config *config.Config) (*ethclient.EthereumClient, er
ethereum := ethclient.NewEthereumClient() ethereum := ethclient.NewEthereumClient()
err := ethereum.Initialize(&config.Ethereum) err := ethereum.Initialize(&config.Ethereum)
if err != nil { if err != nil {
return nil, xerrors.Errorf("failed to initialize ethereum client: %v", err) return nil, err
} }
return ethereum, nil return ethereum, nil
} }
@ -106,7 +106,7 @@ func providePubsubRouter(lhost host.Host, config *config.Config) *pubsub.PubSubR
func provideConsensusManager( func provideConsensusManager(
bus EventBus.Bus, bus EventBus.Bus,
psb *pubsub.PubSubRouter, psb *pubsub.PubSubRouter,
miner *consensus.Miner, miner *blockchain.Miner,
bc *blockchain.BlockChain, bc *blockchain.BlockChain,
ethClient *ethclient.EthereumClient, ethClient *ethclient.EthereumClient,
privateKey crypto.PrivKey, privateKey crypto.PrivKey,
@ -114,6 +114,7 @@ func provideConsensusManager(
bp *pool.BlockPool, bp *pool.BlockPool,
b beacon.BeaconNetwork, b beacon.BeaconNetwork,
mp *pool.Mempool, mp *pool.Mempool,
address peer.ID,
) *consensus.PBFTConsensusManager { ) *consensus.PBFTConsensusManager {
return consensus.NewPBFTConsensusManager( return consensus.NewPBFTConsensusManager(
bus, bus,
@ -126,6 +127,7 @@ func provideConsensusManager(
bp, bp,
b, b,
mp, mp,
address,
) )
} }
@ -176,8 +178,8 @@ func providePeerDiscovery(baddrs []multiaddr.Multiaddr, h host.Host, pexDiscover
return pexDiscovery, nil return pexDiscovery, nil
} }
func provideBlockChain(config *config.Config, bus EventBus.Bus) (*blockchain.BlockChain, error) { func provideBlockChain(config *config.Config, bus EventBus.Bus, miner *blockchain.Miner, b beacon.BeaconAPI) (*blockchain.BlockChain, error) {
return blockchain.NewBlockChain(config.Blockchain.DatabasePath, bus) return blockchain.NewBlockChain(config.Blockchain.DatabasePath, bus, miner, b)
} }
func provideMemPool(bus EventBus.Bus) (*pool.Mempool, error) { func provideMemPool(bus EventBus.Bus) (*pool.Mempool, error) {