[FAB-5433] Move costly checks from the critical path

In gossip, when a message that contains a block is sent to a peer
(or a peer requests a block) - we consult the MessageCryptoService
and ask it- is the given peer (that is considered to receive the
block from us) among the channel readers.

The problem is that this operation involves verifying ECDSA signatures
and it is in the critical path of the dissemination.

This commit moves the policy checks from the critical path to the time
when the stateInfo messages that are used to verify the peers eligibility
are being put into the in-memory cache:
- Whenever the StateInfo message is attempted to be added
  into the in-memory cache, it is checked that the peer
  that signed it- is a channel reader and is in the channel
- Whenver a config updates is received, all messages in the
  in-memory cache are validated, and if some are found to
  be invalid - they are deleted from the cache.

With this commit: whenever the channel module is asked whether a given peer
is eligible of receiving blocks, it simply does a lookup
in the in-memory cache of StateInfo messages.

More details can be found in the JIRA item.

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

Author: yacovm

gossip/gossip/channel/channel.go

@@ -181,7 +181,49 @@ func NewGossipChannel(pkiID common.PKIidType, org api.OrgIdentityType, mcs api.M
 		pkiID := o.(*proto.SignedGossipMessage).GetStateInfo().PkiId
 		return gc.Lookup(pkiID) == nil
 	}
-	gc.stateInfoMsgStore = newStateInfoCache(gc.GetConf().StateInfoCacheSweepInterval, hashPeerExpiredInMembership)
+	verifyStateInfoMsg := func(msg *proto.SignedGossipMessage, orgs ...api.OrgIdentityType) bool {
+		si := msg.GetStateInfo()
+		// No point in verifying ourselves
+		if bytes.Equal(gc.pkiID, si.PkiId) {
+			return true
+		}
+		peerIdentity := adapter.GetIdentityByPKIID(si.PkiId)
+		if len(peerIdentity) == 0 {
+			gc.logger.Warning("Identity for peer", si.PkiId, "doesn't exist")
+			return false
+		}
+		isOrgInChan := func(org api.OrgIdentityType) bool {
+			if len(orgs) == 0 {
+				if !gc.IsOrgInChannel(org) {
+					return false
+				}
+			} else {
+				found := false
+				for _, chanMember := range orgs {
+					if bytes.Equal(chanMember, org) {
+						found = true
+						break
+					}
+				}
+				if !found {
+					return false
+				}
+			}
+			return true
+		}
+
+		org := gc.GetOrgOfPeer(si.PkiId)
+		if !isOrgInChan(org) {
+			gc.logger.Warning("peer", peerIdentity, "'s organization(", string(org), ") isn't in the channel", string(chainID))
+			return false
+		}
+		if err := gc.mcs.VerifyByChannel(chainID, peerIdentity, msg.Signature, msg.Payload); err != nil {
+			gc.logger.Warning("Peer", peerIdentity, "isn't eligible for channel", string(chainID), ":", err)
+			return false
+		}
+		return true
+	}
+	gc.stateInfoMsgStore = newStateInfoCache(gc.GetConf().StateInfoCacheSweepInterval, hashPeerExpiredInMembership, verifyStateInfoMsg)
 
 	ttl := election.GetMsgExpirationTimeout()
 	pol := proto.NewGossipMessageComparator(0)
@@ -321,17 +363,16 @@ func (gc *gossipChannel) IsOrgInChannel(membersOrg api.OrgIdentityType) bool {
 // EligibleForChannel returns whether the given member should get blocks
 // for this channel
 func (gc *gossipChannel) EligibleForChannel(member discovery.NetworkMember) bool {
-	if !gc.IsMemberInChan(member) {
+	peerIdentity := gc.GetIdentityByPKIID(member.PKIid)
+	if len(peerIdentity) == 0 {
+		gc.logger.Warning("Identity for peer", member.PKIid, "doesn't exist")
 		return false
 	}
-
-	identity := gc.GetIdentityByPKIID(member.PKIid)
 	msg := gc.stateInfoMsgStore.MsgByID(member.PKIid)
-	if msg == nil || identity == nil {
+	if msg == nil {
 		return false
 	}
-
-	return gc.mcs.VerifyByChannel(gc.chainID, identity, msg.Envelope.Signature, msg.Envelope.Payload) == nil
+	return true
 }
 
 // AddToMsgStore adds a given GossipMessage to the message store
@@ -368,6 +409,7 @@ func (gc *gossipChannel) ConfigureChannel(joinMsg api.JoinChannelMessage) {
 
 	gc.orgs = joinMsg.Members()
 	gc.joinMsg = joinMsg
+	gc.stateInfoMsgStore.validate(joinMsg.Members())
 }
 
 // HandleMessage processes a message sent by a remote peer
@@ -663,10 +705,12 @@ func (gc *gossipChannel) UpdateStateInfo(msg *proto.SignedGossipMessage) {
 	atomic.StoreInt32(&gc.shouldGossipStateInfo, int32(1))
 }
 
-func newStateInfoCache(sweepInterval time.Duration, hasExpired func(interface{}) bool) *stateInfoCache {
+func newStateInfoCache(sweepInterval time.Duration, hasExpired func(interface{}) bool, verifyFunc membershipPredicate) *stateInfoCache {
 	membershipStore := util.NewMembershipStore()
 	pol := proto.NewGossipMessageComparator(0)
+
 	s := &stateInfoCache{
+		verify:          verifyFunc,
 		MembershipStore: membershipStore,
 		stopChan:        make(chan struct{}),
 	}
@@ -689,19 +733,40 @@ func newStateInfoCache(sweepInterval time.Duration, hasExpired func(interface{})
 	return s
 }
 
+// membershipPredicate receives a StateInfoMessage and optionally a slice of organization identifiers
+// and returns whether the peer that signed the given StateInfoMessage is eligible
+// to the channel or not
+type membershipPredicate func(msg *proto.SignedGossipMessage, orgs ...api.OrgIdentityType) bool
+
 // stateInfoCache is actually a messageStore
 // that also indexes messages that are added
 // so that they could be extracted later
 type stateInfoCache struct {
+	verify membershipPredicate
 	*util.MembershipStore
 	msgstore.MessageStore
 	stopChan chan struct{}
 }
 
+func (cache *stateInfoCache) validate(orgs []api.OrgIdentityType) {
+	for _, m := range cache.Get() {
+		msg := m.(*proto.SignedGossipMessage)
+		if !cache.verify(msg, orgs...) {
+			cache.delete(msg)
+		}
+	}
+}
+
 // Add attempts to add the given message to the stateInfoCache,
 // and if the message was added, also indexes it.
 // Message must be a StateInfo message.
 func (cache *stateInfoCache) Add(msg *proto.SignedGossipMessage) bool {
+	if !cache.MessageStore.CheckValid(msg) {
+		return false
+	}
+	if !cache.verify(msg) {
+		return false
+	}
 	added := cache.MessageStore.Add(msg)
 	if added {
 		pkiID := msg.GetStateInfo().PkiId
@@ -710,6 +775,14 @@ func (cache *stateInfoCache) Add(msg *proto.SignedGossipMessage) bool {
 	return added
 }
 
+func (cache *stateInfoCache) delete(msg *proto.SignedGossipMessage) {
+	cache.Purge(func(o interface{}) bool {
+		pkiID := o.(*proto.SignedGossipMessage).GetStateInfo().PkiId
+		return bytes.Equal(pkiID, msg.GetStateInfo().PkiId)
+	})
+	cache.Remove(msg.GetStateInfo().PkiId)
+}
+
 func (cache *stateInfoCache) Stop() {
 	cache.stopChan <- struct{}{}
 }

gossip/gossip/channel/channel_test.go

@@ -112,6 +112,9 @@ func (cs *cryptoService) VerifyByChannel(channel common.ChainID, identity api.Pe
 		return nil
 	}
 	args := cs.Called(identity)
+	if args.Get(0) == nil {
+		return nil
+	}
 	return args.Get(0).(error)
 }
 
@@ -222,6 +225,9 @@ func (ga *gossipAdapterMock) GetOrgOfPeer(PKIIID common.PKIidType) api.OrgIdenti
 }
 
 func (ga *gossipAdapterMock) GetIdentityByPKIID(pkiID common.PKIidType) api.PeerIdentityType {
+	if ga.wasMocked("GetIdentityByPKIID") {
+		return ga.Called(pkiID).Get(0).(api.PeerIdentityType)
+	}
 	return api.PeerIdentityType(pkiID)
 }
 
@@ -676,10 +682,12 @@ func TestChannelPeerNotInChannel(t *testing.T) {
 
 	// Now for a more advanced scenario- the peer claims to be in the right org, and also claims to be in the channel
 	// but the MSP declares it is not eligible for the channel
-	// pkiIDInOrg1ButNotEligible
 	gc.HandleMessage(&receivedMsg{msg: createStateInfoMsg(10, pkiIDInOrg1ButNotEligible, channelA), PKIID: pkiIDInOrg1ButNotEligible})
+	// configure MSP
 	cs.On("VerifyByChannel", mock.Anything).Return(errors.New("Not eligible"))
 	cs.mocked = true
+	// Simulate a config update
+	gc.ConfigureChannel(&joinChanMsg{})
 	helloMsg = createHelloMsg(pkiIDInOrg1ButNotEligible)
 	helloMsg.On("Respond", mock.Anything).Run(messageRelayer)
 	gc.HandleMessage(helloMsg)
@@ -1400,6 +1408,110 @@ func TestChannelNoAnchorPeers(t *testing.T) {
 	assert.True(t, gc.IsOrgInChannel(orgInChannelA))
 }
 
+func TestGossipChannelEligibility(t *testing.T) {
+	t.Parallel()
+
+	// Scenario: We have a peer in an org that joins a channel with org1 and org2.
+	// and it receives StateInfo messages of other peers and the eligibility
+	// of these peers of being in the channel is checked.
+	// During the test, the channel is reconfigured, and the expiration
+	// of the peer identities is simulated.
+
+	cs := &cryptoService{}
+	selfPKIID := common.PKIidType("p")
+	adapter := new(gossipAdapterMock)
+	pkiIDinOrg3 := common.PKIidType("pkiIDinOrg3")
+	members := []discovery.NetworkMember{
+		{PKIid: pkiIDInOrg1},
+		{PKIid: pkiIDInOrg1ButNotEligible},
+		{PKIid: pkiIDinOrg2},
+	}
+	adapter.On("GetMembership").Return(members)
+	adapter.On("Gossip", mock.Anything)
+	adapter.On("Send", mock.Anything, mock.Anything)
+	adapter.On("DeMultiplex", mock.Anything)
+	adapter.On("GetConf").Return(conf)
+
+	// At first, all peers are in the channel except pkiIDinOrg3
+	org1 := api.OrgIdentityType("ORG1")
+	org2 := api.OrgIdentityType("ORG2")
+	org3 := api.OrgIdentityType("ORG3")
+
+	adapter.On("GetOrgOfPeer", selfPKIID).Return(org1)
+	adapter.On("GetOrgOfPeer", pkiIDInOrg1).Return(org1)
+	adapter.On("GetOrgOfPeer", pkiIDinOrg2).Return(org2)
+	adapter.On("GetOrgOfPeer", pkiIDInOrg1ButNotEligible).Return(org1)
+	adapter.On("GetOrgOfPeer", pkiIDinOrg3).Return(org3)
+
+	gc := NewGossipChannel(selfPKIID, orgInChannelA, cs, channelA, adapter, &joinChanMsg{
+		members2AnchorPeers: map[string][]api.AnchorPeer{
+			string(org1): {},
+			string(org2): {},
+		},
+	})
+	// Every peer sends a StateInfo message
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(1, pkiIDInOrg1, channelA)})
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(1, pkiIDinOrg2, channelA)})
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(1, pkiIDInOrg1ButNotEligible, channelA)})
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(1, pkiIDinOrg3, channelA)})
+
+	assert.True(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1}))
+	assert.True(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg2}))
+	assert.True(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1ButNotEligible}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg3}))
+
+	// Remove org2 from the channel
+	gc.ConfigureChannel(&joinChanMsg{
+		members2AnchorPeers: map[string][]api.AnchorPeer{
+			string(org1): {},
+		},
+	})
+
+	assert.True(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg2}))
+	assert.True(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1ButNotEligible}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg3}))
+
+	// Now simulate a config update that removed pkiIDInOrg1ButNotEligible from the channel readers
+	cs.mocked = true
+	cs.On("VerifyByChannel", api.PeerIdentityType(pkiIDInOrg1ButNotEligible)).Return(errors.New("Not a channel reader"))
+	cs.On("VerifyByChannel", mock.Anything).Return(nil)
+	gc.ConfigureChannel(&joinChanMsg{
+		members2AnchorPeers: map[string][]api.AnchorPeer{
+			string(org1): {},
+		},
+	})
+	assert.True(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg2}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1ButNotEligible}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg3}))
+
+	// Now Simulate a certificate expiration of pkiIDInOrg1.
+	// This is done by asking the adapter to lookup the identity by PKI-ID, but if the certificate
+	// is expired, the mapping is deleted and hence the lookup yields nothing.
+	adapter.On("GetIdentityByPKIID", pkiIDInOrg1).Return(api.PeerIdentityType(nil))
+	adapter.On("GetIdentityByPKIID", pkiIDinOrg2).Return(api.PeerIdentityType(pkiIDinOrg2))
+	adapter.On("GetIdentityByPKIID", pkiIDInOrg1ButNotEligible).Return(api.PeerIdentityType(pkiIDInOrg1ButNotEligible))
+	adapter.On("GetIdentityByPKIID", pkiIDinOrg3).Return(api.PeerIdentityType(pkiIDinOrg3))
+
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg2}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1ButNotEligible}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg3}))
+
+	// Now make another update of StateInfo messages, this time with updated ledger height (to overwrite earlier messages)
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(2, pkiIDInOrg1, channelA)})
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(2, pkiIDinOrg2, channelA)})
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(2, pkiIDInOrg1ButNotEligible, channelA)})
+	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(2, pkiIDinOrg3, channelA)})
+
+	// Ensure the access control resolution hasn't changed
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg2}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDInOrg1ButNotEligible}))
+	assert.False(t, gc.EligibleForChannel(discovery.NetworkMember{PKIid: pkiIDinOrg3}))
+}
+
 func TestChannelGetPeers(t *testing.T) {
 	t.Parallel()
 
@@ -1420,7 +1532,7 @@ func TestChannelGetPeers(t *testing.T) {
 		{PKIid: pkiIDinOrg2},
 	}
 	configureAdapter(adapter, members...)
-	gc := NewGossipChannel(pkiIDInOrg1, orgInChannelA, cs, channelA, adapter, &joinChanMsg{})
+	gc := NewGossipChannel(common.PKIidType("p0"), orgInChannelA, cs, channelA, adapter, &joinChanMsg{})
 	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(1, pkiIDInOrg1, channelA)})
 	gc.HandleMessage(&receivedMsg{PKIID: pkiIDInOrg1, msg: createStateInfoMsg(1, pkiIDinOrg2, channelA)})
 	assert.Len(t, gc.GetPeers(), 1)
@@ -1429,6 +1541,8 @@ func TestChannelGetPeers(t *testing.T) {
 	gc.HandleMessage(&receivedMsg{msg: createStateInfoMsg(10, pkiIDInOrg1ButNotEligible, channelA), PKIID: pkiIDInOrg1ButNotEligible})
 	cs.On("VerifyByChannel", mock.Anything).Return(errors.New("Not eligible"))
 	cs.mocked = true
+	// Simulate a config update
+	gc.ConfigureChannel(&joinChanMsg{})
 	assert.Len(t, gc.GetPeers(), 0)
 
 	// Now recreate gc and corrupt the MAC