[FAB-7554] Include Envelopes in gossip membership API

The discovery service needs to provide clients with envelopes of
peers so that the clients could (optionally) verify the signatures
themselves if they desire.
The discovery.NetworkMember already has an envelope from previous
change sets, and we need to populate the field when we return
the envelopes in the discovery and channel modules.

Change-Id: If582553c4e19d3403c3f8728b3a6fb5a3f8e27ef
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

gossip/discovery/discovery_impl.go

@@ -410,7 +410,7 @@ func (d *gossipDiscoveryImpl) sendMemResponse(targetMember *proto.Member, intern
 		InternalEndpoint: internalEndpoint,
 	}
 
-	aliveMsg, err := d.createAliveMessage(true)
+	aliveMsg, err := d.createSignedAliveMessage(true)
 	if err != nil {
 		d.logger.Warningf("Failed creating alive message: %+v", errors.WithStack(err))
 		return
@@ -624,7 +624,7 @@ func (d *gossipDiscoveryImpl) sendMembershipRequest(member *NetworkMember, inclu
 }
 
 func (d *gossipDiscoveryImpl) createMembershipRequest(includeInternalEndpoint bool) (*proto.GossipMessage, error) {
-	am, err := d.createAliveMessage(includeInternalEndpoint)
+	am, err := d.createSignedAliveMessage(includeInternalEndpoint)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -723,7 +723,7 @@ func (d *gossipDiscoveryImpl) periodicalSendAlive() {
 	for !d.toDie() {
 		d.logger.Debug("Sleeping", getAliveTimeInterval())
 		time.Sleep(getAliveTimeInterval())
-		msg, err := d.createAliveMessage(true)
+		msg, err := d.createSignedAliveMessage(true)
 		if err != nil {
 			d.logger.Warningf("Failed creating alive message: %+v", errors.WithStack(err))
 			return
@@ -732,19 +732,16 @@ func (d *gossipDiscoveryImpl) periodicalSendAlive() {
 	}
 }
 
-func (d *gossipDiscoveryImpl) createAliveMessage(includeInternalEndpoint bool) (*proto.SignedGossipMessage, error) {
+func (d *gossipDiscoveryImpl) aliveMsgAndInternalEndpoint() (*proto.GossipMessage, string) {
 	d.lock.Lock()
+	defer d.lock.Unlock()
 	d.seqNum++
 	seqNum := d.seqNum
-
 	endpoint := d.self.Endpoint
 	meta := d.self.Metadata
 	pkiID := d.self.PKIid
 	internalEndpoint := d.self.InternalEndpoint
-
-	d.lock.Unlock()
-
-	msg2Gossip := &proto.GossipMessage{
+	msg := &proto.GossipMessage{
 		Tag: proto.GossipMessage_EMPTY,
 		Content: &proto.GossipMessage_AliveMsg{
 			AliveMsg: &proto.AliveMessage{
@@ -760,13 +757,17 @@ func (d *gossipDiscoveryImpl) createAliveMessage(includeInternalEndpoint bool) (
 			},
 		},
 	}
+	return msg, internalEndpoint
+}
 
-	envp := d.crypt.SignMessage(msg2Gossip, internalEndpoint)
+func (d *gossipDiscoveryImpl) createSignedAliveMessage(includeInternalEndpoint bool) (*proto.SignedGossipMessage, error) {
+	msg, internalEndpoint := d.aliveMsgAndInternalEndpoint()
+	envp := d.crypt.SignMessage(msg, internalEndpoint)
 	if envp == nil {
 		return nil, errors.New("Failed signing message")
 	}
 	signedMsg := &proto.SignedGossipMessage{
-		GossipMessage: msg2Gossip,
+		GossipMessage: msg,
 		Envelope:      envp,
 	}
 
@@ -913,6 +914,7 @@ func (d *gossipDiscoveryImpl) GetMembership() []NetworkMember {
 			Endpoint:         member.Membership.Endpoint,
 			Metadata:         member.Membership.Metadata,
 			InternalEndpoint: d.id2Member[string(m.GetAliveMsg().Membership.PkiId)].InternalEndpoint,
+			Envelope:         m.Envelope,
 		})
 	}
 	return response
@@ -937,11 +939,20 @@ func (d *gossipDiscoveryImpl) UpdateEndpoint(endpoint string) {
 }
 
 func (d *gossipDiscoveryImpl) Self() NetworkMember {
+	var env *proto.Envelope
+	msg, _ := d.aliveMsgAndInternalEndpoint()
+	sMsg, err := msg.NoopSign()
+	if err != nil {
+		d.logger.Warning("Failed creating SignedGossipMessage:", err)
+	} else {
+		env = sMsg.Envelope
+	}
+	mem := msg.GetAliveMsg().Membership
 	return NetworkMember{
-		Endpoint:         d.self.Endpoint,
-		Metadata:         d.self.Metadata,
-		PKIid:            d.self.PKIid,
-		InternalEndpoint: d.self.InternalEndpoint,
+		Endpoint: mem.Endpoint,
+		Metadata: mem.Metadata,
+		PKIid:    mem.PkiId,
+		Envelope: env,
 	}
 }
 

gossip/discovery/discovery_test.go

@@ -545,6 +545,21 @@ func TestInitiateSync(t *testing.T) {
 	stopInstances(t, instances)
 }
 
+func TestSelf(t *testing.T) {
+	t.Parallel()
+	inst := createDiscoveryInstance(13463, "d1", []string{})
+	defer inst.Stop()
+	env := inst.Self().Envelope
+	sMsg, err := env.ToGossipMessage()
+	assert.NoError(t, err)
+	member := sMsg.GetAliveMsg().Membership
+	assert.Equal(t, "localhost:13463", member.Endpoint)
+	assert.Equal(t, []byte("localhost:13463"), member.PkiId)
+
+	assert.Equal(t, "localhost:13463", inst.Self().Endpoint)
+	assert.Equal(t, common.PKIidType("localhost:13463"), inst.Self().PKIid)
+}
+
 func TestExpiration(t *testing.T) {
 	t.Parallel()
 	nodeNum := 5
@@ -974,7 +989,7 @@ func TestMsgStoreExpirationWithMembershipMessages(t *testing.T) {
 	}
 	// Creating Alive messages
 	for i := 0; i < peersNum; i++ {
-		aliveMsg, _ := instances[i].discoveryImpl().createAliveMessage(true)
+		aliveMsg, _ := instances[i].discoveryImpl().createSignedAliveMessage(true)
 		aliveMsgs = append(aliveMsgs, aliveMsg)
 	}
 
@@ -1042,15 +1057,15 @@ func TestMsgStoreExpirationWithMembershipMessages(t *testing.T) {
 				return k == i
 			},
 			func(k int) {
-				aliveMsg, _ := instances[k].discoveryImpl().createAliveMessage(true)
+				aliveMsg, _ := instances[k].discoveryImpl().createSignedAliveMessage(true)
 				memResp := instances[k].discoveryImpl().createMembershipResponse(aliveMsg, peerToResponse)
 				memRespMsgs[i] = append(memRespMsgs[i], memResp)
 			})
 	}
 
 	// Re-creating Alive msgs with highest seq_num, to make sure Alive msgs in memReq and memResp are older
 	for i := 0; i < peersNum; i++ {
-		aliveMsg, _ := instances[i].discoveryImpl().createAliveMessage(true)
+		aliveMsg, _ := instances[i].discoveryImpl().createSignedAliveMessage(true)
 		newAliveMsgs = append(newAliveMsgs, aliveMsg)
 	}
 
@@ -1183,7 +1198,7 @@ func TestAliveMsgStore(t *testing.T) {
 	}
 	// Creating Alive messages
 	for i := 0; i < peersNum; i++ {
-		aliveMsg, _ := instances[i].discoveryImpl().createAliveMessage(true)
+		aliveMsg, _ := instances[i].discoveryImpl().createSignedAliveMessage(true)
 		aliveMsgs = append(aliveMsgs, aliveMsg)
 	}
 

gossip/gossip/channel/channel.go

@@ -45,6 +45,9 @@ type Config struct {
 // GossipChannel defines an object that deals with all channel-related messages
 type GossipChannel interface {
 
+	// Self returns a StateInfoMessage about the peer
+	Self() *proto.SignedGossipMessage
+
 	// GetPeers returns a list of peers with metadata as published by them
 	GetPeers() []discovery.NetworkMember
 
@@ -287,6 +290,13 @@ func (gc *gossipChannel) periodicalInvocation(fn func(), c <-chan time.Time) {
 	}
 }
 
+// Self returns a StateInfoMessage about the peer
+func (gc *gossipChannel) Self() *proto.SignedGossipMessage {
+	gc.RLock()
+	defer gc.RUnlock()
+	return gc.stateInfoMsg
+}
+
 // LeaveChannel makes the peer leave the channel
 func (gc *gossipChannel) LeaveChannel() {
 	gc.Lock()
@@ -309,7 +319,7 @@ func (gc *gossipChannel) hasLeftChannel() bool {
 
 // GetPeers returns a list of peers with metadata as published by them
 func (gc *gossipChannel) GetPeers() []discovery.NetworkMember {
-	members := []discovery.NetworkMember{}
+	var members []discovery.NetworkMember
 	if gc.hasLeftChannel() {
 		return members
 	}
@@ -327,6 +337,7 @@ func (gc *gossipChannel) GetPeers() []discovery.NetworkMember {
 			continue
 		}
 		member.Properties = stateInf.GetStateInfo().Properties
+		member.Envelope = stateInf.Envelope
 		members = append(members, member)
 	}
 	return members

gossip/gossip/channel/channel_test.go

@@ -282,6 +282,29 @@ func TestBadInput(t *testing.T) {
 	assert.False(t, gc.verifyMsg(&receivedMsg{msg: nil, PKIID: nil}))
 }
 
+func TestSelf(t *testing.T) {
+	t.Parallel()
+
+	cs := &cryptoService{}
+	pkiID1 := common.PKIidType("1")
+	jcm := &joinChanMsg{
+		members2AnchorPeers: map[string][]api.AnchorPeer{
+			string(orgInChannelA): {},
+		},
+	}
+	adapter := new(gossipAdapterMock)
+	configureAdapter(adapter)
+	adapter.On("Gossip", mock.Anything)
+	gc := NewGossipChannel(pkiID1, orgInChannelA, cs, channelA, adapter, jcm)
+	gc.UpdateLedgerHeight(1)
+	gMsg := gc.Self().GossipMessage
+	env := gc.Self().Envelope
+	sMsg, _ := env.ToGossipMessage()
+	assert.Equal(t, gMsg, sMsg.GossipMessage)
+	assert.Equal(t, gMsg.GetStateInfo().Properties.LedgerHeight, uint64(1))
+	assert.Equal(t, gMsg.GetStateInfo().PkiId, []byte("1"))
+}
+
 func TestMsgStoreNotExpire(t *testing.T) {
 	t.Parallel()
 
@@ -1660,6 +1683,10 @@ func TestChannelGetPeers(t *testing.T) {
 	assert.Len(t, gc.GetPeers(), 1)
 	assert.Equal(t, pkiIDInOrg1, gc.GetPeers()[0].PKIid)
 
+	// Ensure envelope from GetPeers is valid
+	gMsg, _ := gc.GetPeers()[0].Envelope.ToGossipMessage()
+	assert.Equal(t, []byte(pkiIDInOrg1), gMsg.GetStateInfo().PkiId)
+
 	gc.HandleMessage(&receivedMsg{msg: createStateInfoMsg(10, pkiIDInOrg1ButNotEligible, channelA), PKIID: pkiIDInOrg1ButNotEligible})
 	cs.On("VerifyByChannel", mock.Anything).Return(errors.New("Not eligible"))
 	cs.mocked = true

gossip/gossip/gossip.go

@@ -21,6 +21,12 @@ import (
 // Gossip is the interface of the gossip component
 type Gossip interface {
 
+	// SelfMembershipInfo returns the peer's membership information
+	SelfMembershipInfo() discovery.NetworkMember
+
+	// SelfChannelInfo returns the peer's latest StateInfo message of a given channel
+	SelfChannelInfo(common.ChainID) *proto.SignedGossipMessage
+
 	// Send sends a message to remote peers
 	Send(msg *proto.GossipMessage, peers ...*comm.RemotePeer)
 

gossip/gossip/gossip_impl.go

@@ -727,6 +727,20 @@ func (g *gossipServiceImpl) PeersOfChannel(channel common.ChainID) []discovery.N
 	return gc.GetPeers()
 }
 
+// SelfMembershipInfo returns the peer's membership information
+func (g *gossipServiceImpl) SelfMembershipInfo() discovery.NetworkMember {
+	return g.disc.Self()
+}
+
+// SelfChannelInfo returns the peer's latest StateInfo message of a given channel
+func (g *gossipServiceImpl) SelfChannelInfo(chain common.ChainID) *proto.SignedGossipMessage {
+	ch := g.chanState.getGossipChannelByChainID(chain)
+	if ch == nil {
+		return nil
+	}
+	return ch.Self()
+}
+
 // PeerFilter receives a SubChannelSelectionCriteria and returns a RoutingFilter that selects
 // only peer identities that match the given criteria, and that they published their channel participation
 func (g *gossipServiceImpl) PeerFilter(channel common.ChainID, messagePredicate api.SubChannelSelectionCriteria) (filter.RoutingFilter, error) {

gossip/service/join_test.go

@@ -40,6 +40,14 @@ type gossipMock struct {
 	mock.Mock
 }
 
+func (g *gossipMock) SelfChannelInfo(common.ChainID) *proto.SignedGossipMessage {
+	panic("implement me")
+}
+
+func (g *gossipMock) SelfMembershipInfo() discovery.NetworkMember {
+	panic("implement me")
+}
+
 func (*gossipMock) PeerFilter(channel common.ChainID, messagePredicate api.SubChannelSelectionCriteria) (filter.RoutingFilter, error) {
 	panic("implement me")
 }

gossip/state/mocks/gossip.go

@@ -21,6 +21,14 @@ type GossipMock struct {
 	mock.Mock
 }
 
+func (g *GossipMock) SelfMembershipInfo() discovery.NetworkMember {
+	panic("implement me")
+}
+
+func (g *GossipMock) SelfChannelInfo(common.ChainID) *proto.SignedGossipMessage {
+	panic("implement me")
+}
+
 func (*GossipMock) PeerFilter(channel common.ChainID, messagePredicate api.SubChannelSelectionCriteria) (filter.RoutingFilter, error) {
 	panic("implement me")
 }