Add init nad commit message construct/process/passing

consensus/consensus.go

@@ -218,8 +218,7 @@ func New(host p2p.Host, ShardID string, peers []p2p.Peer, leader p2p.Peer) *Cons
 	return &consensus
 }
 
-// Checks the basic meta of a consensus message.
-//
+// Checks the basic meta of a consensus message, including the signature.
 func (consensus *Consensus) checkConsensusMessage(message consensus_proto.Message, publicKey *bls.PublicKey) error {
 	consensusID := message.ConsensusId
 	blockHash := message.BlockHash

drand/drand.go

@@ -3,6 +3,7 @@ package drand
 import (
 	"crypto/sha256"
 	"encoding/binary"
+	"errors"
 	"strconv"
 	"sync"
 
@@ -16,7 +17,7 @@ import (
 
 // DRand is the main struct which contains state for the distributed randomness protocol.
 type DRand struct {
-	vrfs   *map[uint32][32]byte
+	vrfs   *map[uint32][]byte
 	bitmap *bls_cosi.Mask
 	pRand  *[32]byte
 	rand   *[32]byte
@@ -68,7 +69,7 @@ func New(host p2p.Host, ShardID string, peers []p2p.Peer, leader p2p.Peer) *DRan
 		dRand.validators.Store(utils.GetUniqueIDFromPeer(peer), peer)
 	}
 
-	dRand.vrfs = &map[uint32][32]byte{}
+	dRand.vrfs = &map[uint32][]byte{}
 
 	// Initialize cosign bitmap
 	allPublicKeys := make([]*bls.PublicKey, 0)
@@ -136,7 +137,7 @@ func (dRand *DRand) signAndMarshalDRandMessage(message *drand_proto.Message) ([]
 	return marshaledMessage, nil
 }
 
-func (dRand *DRand) vrf() (rand [32]byte, proof []byte) {
+func (dRand *DRand) vrf(blockHash [32]byte) (rand [32]byte, proof []byte) {
 	// TODO: implement vrf
 	return [32]byte{}, []byte{}
 }
@@ -155,3 +156,39 @@ func (dRand *DRand) GetValidatorPeers() []p2p.Peer {
 
 	return validatorPeers
 }
+
+// Verify the signature of the message are valid from the signer's public key.
+func verifyMessageSig(signerPubKey *bls.PublicKey, message drand_proto.Message) error {
+	signature := message.Signature
+	message.Signature = nil
+	messageBytes, err := protobuf.Marshal(&message)
+	if err != nil {
+		return err
+	}
+
+	msgSig := bls.Sign{}
+	err = msgSig.Deserialize(signature)
+	if err != nil {
+		return err
+	}
+	msgHash := sha256.Sum256(messageBytes)
+	if !msgSig.VerifyHash(signerPubKey, msgHash[:]) {
+		return errors.New("failed to verify the signature")
+	}
+	return nil
+}
+
+// Gets the validator peer based on validator ID.
+func (dRand *DRand) getValidatorPeerByID(validatorID uint32) *p2p.Peer {
+	v, ok := dRand.validators.Load(validatorID)
+	if !ok {
+		utils.GetLogInstance().Warn("Unrecognized validator", "validatorID", validatorID, "dRand", dRand)
+		return nil
+	}
+	value, ok := v.(p2p.Peer)
+	if !ok {
+		utils.GetLogInstance().Warn("Invalid validator", "validatorID", validatorID, "dRand", dRand)
+		return nil
+	}
+	return &value
+}

drand/drand_leader.go

@@ -1,7 +1,10 @@
 package drand
 
 import (
+	protobuf "github.com/golang/protobuf/proto"
+	drand_proto "github.com/harmony-one/harmony/api/drand"
 	"github.com/harmony-one/harmony/core/types"
+	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/harmony-one/harmony/p2p/host"
 )
 
@@ -32,7 +35,59 @@ func (dRand *DRand) init(epochBlock *types.Block) {
 	msgToSend := dRand.constructInitMessage()
 
 	// Leader commit vrf itself
-	(*dRand.vrfs)[dRand.nodeID], _ = dRand.vrf()
+	rand, proof := dRand.vrf(dRand.blockHash)
+
+	(*dRand.vrfs)[dRand.nodeID] = append(rand[:], proof...)
 
 	host.BroadcastMessageFromLeader(dRand.host, dRand.GetValidatorPeers(), msgToSend, nil)
 }
+
+// ProcessMessageLeader dispatches messages for the leader to corresponding processors.
+func (dRand *DRand) ProcessMessageLeader(payload []byte) {
+	message := drand_proto.Message{}
+	err := protobuf.Unmarshal(payload, &message)
+
+	if err != nil {
+		utils.GetLogInstance().Error("Failed to unmarshal message payload.", "err", err, "dRand", dRand)
+	}
+
+	switch message.Type {
+	case drand_proto.MessageType_COMMIT:
+		dRand.processCommitMessage(message)
+	default:
+		utils.GetLogInstance().Error("Unexpected message type", "msgType", message.Type, "dRand", dRand)
+	}
+}
+
+// ProcessMessageValidator dispatches validator's consensus message.
+func (dRand *DRand) processCommitMessage(message drand_proto.Message) {
+	if message.Type != drand_proto.MessageType_COMMIT {
+		utils.GetLogInstance().Error("Wrong message type received", "expected", drand_proto.MessageType_COMMIT, "got", message.Type)
+		return
+	}
+
+	// Verify message signature
+	err := verifyMessageSig(dRand.leader.PubKey, message)
+	if err != nil {
+		utils.GetLogInstance().Warn("Failed to verify the message signature", "Error", err)
+		return
+	}
+
+	rand := message.Payload[:32]
+	proof := message.Payload[32:]
+	_ = rand
+	_ = proof
+	// TODO: check the validity of the vrf commit
+
+	validatorID := message.SenderId
+	validatorPeer := dRand.getValidatorPeerByID(validatorID)
+	vrfs := dRand.vrfs
+	utils.GetLogInstance().Debug("Received new prepare signature", "numReceivedSoFar", len((*vrfs)), "validatorID", validatorID, "PublicKeys", len(dRand.PublicKeys))
+
+	(*vrfs)[validatorID] = message.Payload
+	dRand.bitmap.SetKey(validatorPeer.PubKey, true) // Set the bitmap indicating that this validator signed.
+
+	if len((*vrfs)) >= ((len(dRand.PublicKeys))/3 + 1) {
+		// Construct pRand and initiate consensus on it
+	}
+}

drand/drand_leader_msg.go

@@ -11,7 +11,7 @@ func (drand *DRand) constructInitMessage() []byte {
 	message := drand_proto.Message{}
 	message.Type = drand_proto.MessageType_INIT
 
-	copy(message.BlockHash, drand.blockHash[:])
+	message.BlockHash = drand.blockHash[:]
 	// Don't need the payload in init message
 	marshaledMessage, err := drand.signAndMarshalDRandMessage(&message)
 	if err != nil {

drand/drand_leader_msg_test.go

@@ -20,7 +20,7 @@ func TestConstructInitMessage(test *testing.T) {
 	dRand.blockHash = [32]byte{}
 	msg := dRand.constructInitMessage()
 
-	if len(msg) != 53 {
+	if len(msg) != 87 {
 		test.Errorf("Init message is not constructed in the correct size: %d", len(msg))
 	}
 }

drand/drand_validator.go

@@ -0,0 +1,52 @@
+package drand
+
+import (
+	protobuf "github.com/golang/protobuf/proto"
+	drand_proto "github.com/harmony-one/harmony/api/drand"
+	"github.com/harmony-one/harmony/internal/utils"
+	"github.com/harmony-one/harmony/p2p/host"
+)
+
+// ProcessMessageValidator dispatches messages for the validator to corresponding processors.
+func (dRand *DRand) ProcessMessageValidator(payload []byte) {
+	message := drand_proto.Message{}
+	err := protobuf.Unmarshal(payload, &message)
+
+	if err != nil {
+		utils.GetLogInstance().Error("Failed to unmarshal message payload.", "err", err, "dRand", dRand)
+	}
+
+	switch message.Type {
+	case drand_proto.MessageType_COMMIT:
+		dRand.processInitMessage(message)
+	default:
+		utils.GetLogInstance().Error("Unexpected message type", "msgType", message.Type, "dRand", dRand)
+	}
+}
+
+// ProcessMessageValidator dispatches validator's consensus message.
+func (dRand *DRand) processInitMessage(message drand_proto.Message) {
+	if message.Type != drand_proto.MessageType_INIT {
+		utils.GetLogInstance().Error("Wrong message type received", "expected", drand_proto.MessageType_INIT, "got", message.Type)
+		return
+	}
+
+	blockHash := message.BlockHash
+
+	// Verify message signature
+	err := verifyMessageSig(dRand.leader.PubKey, message)
+	if err != nil {
+		utils.GetLogInstance().Warn("Failed to verify the message signature", "Error", err)
+		return
+	}
+
+	// TODO: check the blockHash is the block hash of last block of last epoch.
+	copy(dRand.blockHash[:], blockHash[:])
+
+	rand, proof := dRand.vrf(dRand.blockHash)
+
+	msgToSend := dRand.constructCommitMessage(rand, proof)
+
+	// Send the commit message back to leader
+	host.SendMessage(dRand.host, dRand.leader, msgToSend, nil)
+}

drand/drand_validator_msg.go

@@ -0,0 +1,22 @@
+package drand
+
+import (
+	drand_proto "github.com/harmony-one/harmony/api/drand"
+	"github.com/harmony-one/harmony/api/proto"
+	"github.com/harmony-one/harmony/internal/utils"
+)
+
+// Constructs the init message
+func (drand *DRand) constructCommitMessage(vrf [32]byte, proof []byte) []byte {
+	message := drand_proto.Message{}
+	message.Type = drand_proto.MessageType_COMMIT
+
+	message.BlockHash = drand.blockHash[:]
+	message.Payload = append(vrf[:], proof...)
+
+	marshaledMessage, err := drand.signAndMarshalDRandMessage(&message)
+	if err != nil {
+		utils.GetLogInstance().Error("Failed to sign and marshal the commit message", "error", err)
+	}
+	return proto.ConstructDRandMessage(marshaledMessage)
+}

drand/drand_validator_msg_test.go

@@ -0,0 +1,26 @@
+package drand
+
+import (
+	"testing"
+
+	"github.com/harmony-one/harmony/internal/utils"
+	"github.com/harmony-one/harmony/p2p"
+	"github.com/harmony-one/harmony/p2p/p2pimpl"
+)
+
+func TestConstructCommitMessage(test *testing.T) {
+	leader := p2p.Peer{IP: "127.0.0.1", Port: "19999"}
+	validator := p2p.Peer{IP: "127.0.0.1", Port: "55555"}
+	priKey, _, _ := utils.GenKeyP2P("127.0.0.1", "9902")
+	host, err := p2pimpl.NewHost(&leader, priKey)
+	if err != nil {
+		test.Fatalf("newhost failure: %v", err)
+	}
+	dRand := New(host, "0", []p2p.Peer{leader, validator}, leader)
+	dRand.blockHash = [32]byte{}
+	msg := dRand.constructCommitMessage([32]byte{}, []byte{})
+
+	if len(msg) != 121 {
+		test.Errorf("Commit message is not constructed in the correct size: %d", len(msg))
+	}
+}

test/chain/main.go

@@ -96,7 +96,7 @@ func main() {
 		txs[i] = tx
 	}
 	//Add a contract deployment transaction.
-	contractData := "0x60806040526802b5e3af16b188000060015560028054600160a060020a031916331790556101aa806100326000396000f3fe608060405260043610610045577c0100000000000000000000000000000000000000000000000000000000600035046327c78c42811461004a5780634ddd108a1461008c575b600080fd5b34801561005657600080fd5b5061008a6004803603602081101561006d57600080fd5b503573ffffffffffffffffffffffffffffffffffffffff166100b3565b005b34801561009857600080fd5b506100a1610179565b60408051918252519081900360200190f35b60025473ffffffffffffffffffffffffffffffffffffffff1633146100d757600080fd5b600154303110156100e757600080fd5b73ffffffffffffffffffffffffffffffffffffffff811660009081526020819052604090205460ff161561011a57600080fd5b73ffffffffffffffffffffffffffffffffffffffff8116600081815260208190526040808220805460ff1916600190811790915554905181156108fc0292818181858888f19350505050158015610175573d6000803e3d6000fd5b5050565b30319056fea165627a7a7230582003d799bcee73e96e0f40ca432d9c3d2aa9c00a1eba8d00877114a0d7234790ce0029"
+	contractData := "0x60806040526706f05b59d3b2000060015560028054600160a060020a031916331790556101aa806100316000396000f3fe608060405260043610610045577c0100000000000000000000000000000000000000000000000000000000600035046327c78c42811461004a578063b69ef8a81461008c575b600080fd5b34801561005657600080fd5b5061008a6004803603602081101561006d57600080fd5b503573ffffffffffffffffffffffffffffffffffffffff166100b3565b005b34801561009857600080fd5b506100a1610179565b60408051918252519081900360200190f35b60025473ffffffffffffffffffffffffffffffffffffffff1633146100d757600080fd5b600154303110156100e757600080fd5b73ffffffffffffffffffffffffffffffffffffffff811660009081526020819052604090205460ff161561011a57600080fd5b73ffffffffffffffffffffffffffffffffffffffff8116600081815260208190526040808220805460ff1916600190811790915554905181156108fc0292818181858888f19350505050158015610175573d6000803e3d6000fd5b5050565b30319056fea165627a7a723058206b894c1f3badf3b26a7a2768ab8141b1e6fa1c1ddc4622f4f44a7d5041edc9350029"
 	_ = contractData
 	dataEnc := common.FromHex(contractData)