[FAB-5752] Gossip identity expiration III

This commit:
1) Makes gossip's identity package use the new method
   Expiration() that was added to the MessageCryptoService
   in the previous commit.
   Now, when an identity is added to the identity mapper -
   a timer is set and the identity is automatically purged
   after the timer expires.
   The timer is set to the time left until the identity
   is expired.
2) Adds some tests to the identity package, and makes its
   code-coverage reach 100%.
3) Moves the logic that periodically purges un-used identities
   from the gossip package to the identity package, to improve
   code cohesion.

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

Author: yacovm

gossip/gossip/certstore.go

@@ -133,14 +133,11 @@ func (cs *certStore) createIdentityMessage() (*proto.SignedGossipMessage, error)
 	return sMsg, errors.WithStack(err)
 }
 
-func (cs *certStore) listRevokedPeers(isSuspected api.PeerSuspector) []common.PKIidType {
-	revokedPeers := cs.idMapper.ListInvalidIdentities(isSuspected)
-	for _, pkiID := range revokedPeers {
-		cs.pull.Remove(string(pkiID))
-	}
-	return revokedPeers
+func (cs *certStore) suspectPeers(isSuspected api.PeerSuspector) {
+	cs.idMapper.SuspectPeers(isSuspected)
 }
 
 func (cs *certStore) stop() {
 	cs.pull.Stop()
+	cs.idMapper.Stop()
 }

gossip/gossip/certstore_test.go

@@ -89,6 +89,7 @@ func TestCertStoreBadSignature(t *testing.T) {
 	}
 	pm, cs, _ := createObjects(badSignature, nil)
 	defer pm.Stop()
+	defer cs.stop()
 	testCertificateUpdate(t, false, cs)
 }
 
@@ -99,6 +100,7 @@ func TestCertStoreMismatchedIdentity(t *testing.T) {
 
 	pm, cs, _ := createObjects(mismatchedIdentity, nil)
 	defer pm.Stop()
+	defer cs.stop()
 	testCertificateUpdate(t, false, cs)
 }
 
@@ -109,18 +111,15 @@ func TestCertStoreShouldSucceed(t *testing.T) {
 
 	pm, cs, _ := createObjects(totallyFineIdentity, nil)
 	defer pm.Stop()
+	defer cs.stop()
 	testCertificateUpdate(t, true, cs)
 }
 
 func TestCertRevocation(t *testing.T) {
-	identityExpCheckInterval := identityExpirationCheckInterval
 	defer func() {
-		identityExpirationCheckInterval = identityExpCheckInterval
 		cs.revokedPkiIDS = map[string]struct{}{}
 	}()
 
-	identityExpirationCheckInterval = time.Second
-
 	totallyFineIdentity := func(nonce uint64) proto.ReceivedMessage {
 		return createUpdateMessage(nonce, createValidUpdateMessage())
 	}
@@ -130,6 +129,7 @@ func TestCertRevocation(t *testing.T) {
 	pm, cStore, sender := createObjects(totallyFineIdentity, func(message *proto.SignedGossipMessage) {
 		askedForIdentity <- struct{}{}
 	})
+	defer cStore.stop()
 	defer pm.Stop()
 	testCertificateUpdate(t, true, cStore)
 	// Should have asked for an identity for the first time
@@ -172,12 +172,12 @@ func TestCertRevocation(t *testing.T) {
 	select {
 	case <-time.After(time.Second * 5):
 	case <-askedForIdentity:
-		assert.Fail(t, "Shouldn't have asked for an identity, becase we already have it")
+		assert.Fail(t, "Shouldn't have asked for an identity, because we already have it")
 	}
 	assert.Len(t, askedForIdentity, 0)
 	// Revoke the identity
 	cs.revoke(common.PKIidType("B"))
-	cStore.listRevokedPeers(func(id api.PeerIdentityType) bool {
+	cStore.suspectPeers(func(id api.PeerIdentityType) bool {
 		return string(id) == "B"
 	})
 
@@ -215,11 +215,11 @@ func TestCertExpiration(t *testing.T) {
 	defer identity.SetIdentityUsageThreshold(idUsageThreshold)
 
 	// Backup original identityInactivityCheckInterval value
-	inactivityCheckInterval := identityInactivityCheckInterval
-	identityInactivityCheckInterval = time.Second * 1
+	usageThreshold := identity.GetIdentityUsageThreshold()
+	identity.SetIdentityUsageThreshold(time.Second)
 	// Restore original identityInactivityCheckInterval value
 	defer func() {
-		identityInactivityCheckInterval = inactivityCheckInterval
+		identity.SetIdentityUsageThreshold(usageThreshold)
 	}()
 
 	g1 := newGossipInstance(4321, 0, 0, 1)
@@ -430,7 +430,9 @@ func createObjects(updateFactory func(uint64) proto.ReceivedMessage, msgCons pro
 	selfIdentity := api.PeerIdentityType("SELF")
 	certStore = newCertStore(&pullerMock{
 		Mediator: pullMediator,
-	}, identity.NewIdentityMapper(cs, selfIdentity, func(_ common.PKIidType, _ api.PeerIdentityType) {}), selfIdentity, cs)
+	}, identity.NewIdentityMapper(cs, selfIdentity, func(pkiID common.PKIidType, _ api.PeerIdentityType) {
+		pullMediator.Remove(string(pkiID))
+	}), selfIdentity, cs)
 
 	wg := sync.WaitGroup{}
 	wg.Add(1)

gossip/gossip/gossip_impl.go

@@ -36,11 +36,6 @@ const (
 	acceptChanSize       = 100
 )
 
-var (
-	identityExpirationCheckInterval = time.Hour * 24
-	identityInactivityCheckInterval = time.Minute * 10
-)
-
 type channelRoutingFilterFactory func(channel.GossipChannel) filter.RoutingFilter
 
 type gossipServiceImpl struct {
@@ -95,6 +90,7 @@ func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvis
 	g.stateInfoMsgStore = g.newStateInfoMsgStore()
 
 	g.idMapper = identity.NewIdentityMapper(mcs, selfIdentity, func(pkiID common.PKIidType, identity api.PeerIdentityType) {
+		g.comm.CloseConn(&comm.RemotePeer{PKIID: pkiID})
 		g.certPuller.Remove(string(pkiID))
 	})
 
@@ -127,7 +123,6 @@ func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvis
 	}
 
 	go g.start()
-	go g.periodicalIdentityValidationAndExpiration()
 	go g.connect2BootstrapPeers()
 
 	return g
@@ -196,24 +191,7 @@ func (g *gossipServiceImpl) JoinChan(joinMsg api.JoinChannelMessage, chainID com
 // SuspectPeers makes the gossip instance validate identities of suspected peers, and close
 // any connections to peers with identities that are found invalid
 func (g *gossipServiceImpl) SuspectPeers(isSuspected api.PeerSuspector) {
-	for _, pkiID := range g.certStore.listRevokedPeers(isSuspected) {
-		g.comm.CloseConn(&comm.RemotePeer{PKIID: pkiID})
-	}
-}
-
-func (g *gossipServiceImpl) periodicalIdentityValidationAndExpiration() {
-	// We check once every identityExpirationCheckInterval for identities that have been expired
-	go g.periodicalIdentityValidation(func(identity api.PeerIdentityType) bool {
-		// We need to validate every identity to check if it has been expired
-		return true
-	}, identityExpirationCheckInterval)
-
-	// We check once every identityInactivityCheckInterval for identities that have not been used for a long time
-	go g.periodicalIdentityValidation(func(identity api.PeerIdentityType) bool {
-		// We don't validate any identity, because we just want to know whether
-		// it has not been used for a long time
-		return false
-	}, identityInactivityCheckInterval)
+	g.certStore.suspectPeers(isSuspected)
 }
 
 func (g *gossipServiceImpl) periodicalIdentityValidation(suspectFunc api.PeerSuspector, interval time.Duration) {

gossip/gossip/gossip_test.go

@@ -63,9 +63,10 @@ func init() {
 		testWG.Add(1)
 	}
 	factory.InitFactories(nil)
-	identityExpirationCheckInterval = time.Second
 }
 
+var expirationTimes map[string]time.Time = map[string]time.Time{}
+
 var orgInChannelA = api.OrgIdentityType("ORG1")
 
 func acceptData(m interface{}) bool {
@@ -120,6 +121,13 @@ type naiveCryptoService struct {
 	revokedPkiIDS map[string]struct{}
 }
 
+func (*naiveCryptoService) Expiration(peerIdentity api.PeerIdentityType) (time.Time, error) {
+	if exp, exists := expirationTimes[string(peerIdentity)]; exists {
+		return exp, nil
+	}
+	return time.Now().Add(time.Hour), nil
+}
+
 type orgCryptoService struct {
 }
 
@@ -147,10 +155,6 @@ func (cs *naiveCryptoService) VerifyByChannel(_ common.ChainID, identity api.Pee
 	return errors.New("Forbidden")
 }
 
-func (*naiveCryptoService) Expiration(peerIdentity api.PeerIdentityType) (time.Time, error) {
-	return time.Time{}, nil
-}
-
 func (cs *naiveCryptoService) ValidateIdentity(peerIdentity api.PeerIdentityType) error {
 	cs.RLock()
 	defer cs.RUnlock()
@@ -1004,18 +1008,30 @@ func TestDisseminateAll2All(t *testing.T) {
 func TestIdentityExpiration(t *testing.T) {
 	t.Parallel()
 	defer testWG.Done()
-	// Scenario: spawn 4 peers and make the MessageCryptoService revoke one of them.
+	// Scenario: spawn 5 peers and make the MessageCryptoService revoke one of the first 4.
+	// The last peer's certificate expires after a few seconds.
 	// Eventually, the rest of the peers should not be able to communicate with
 	// the revoked peer at all because its identity would seem to them as expired
 
+	// Set expiration of the last peer to 5 seconds from now
+	expirationTimes["localhost:7004"] = time.Now().Add(time.Second * 5)
+
 	portPrefix := 7000
 	g1 := newGossipInstance(portPrefix, 0, 100)
 	g2 := newGossipInstance(portPrefix, 1, 100, 0)
 	g3 := newGossipInstance(portPrefix, 2, 100, 0)
 	g4 := newGossipInstance(portPrefix, 3, 100, 0)
+	g5 := newGossipInstance(portPrefix, 4, 100, 0)
 
 	peers := []Gossip{g1, g2, g3, g4}
 
+	// Make the last peer be revoked in 5 seconds from now
+	time.AfterFunc(time.Second*5, func() {
+		for _, p := range peers {
+			p.(*gossipServiceImpl).mcs.(*naiveCryptoService).revoke(common.PKIidType("localhost:7004"))
+		}
+	})
+
 	seeAllNeighbors := func() bool {
 		for i := 0; i < 4; i++ {
 			neighborCount := len(peers[i].Peers())
@@ -1035,22 +1051,35 @@ func TestIdentityExpiration(t *testing.T) {
 		}
 		p.(*gossipServiceImpl).mcs.(*naiveCryptoService).revoke(revokedPkiID)
 	}
+	// Trigger a config update to the rest of the peers
+	for i := 0; i < 4; i++ {
+		if i == revokedPeerIndex {
+			continue
+		}
+		peers[i].SuspectPeers(func(_ api.PeerIdentityType) bool {
+			return true
+		})
+	}
 	// Ensure that no one talks to the peer that is revoked
 	ensureRevokedPeerIsIgnored := func() bool {
 		for i := 0; i < 4; i++ {
 			neighborCount := len(peers[i].Peers())
 			expectedNeighborCount := 2
-			if i == revokedPeerIndex {
+			// If it's the revoked peer, or the last peer who's certificate
+			// has expired
+			if i == revokedPeerIndex || i == 4 {
 				expectedNeighborCount = 0
 			}
 			if neighborCount != expectedNeighborCount {
+				fmt.Println("neighbor count of", i, "is", neighborCount)
 				return false
 			}
 		}
 		return true
 	}
 	waitUntilOrFail(t, ensureRevokedPeerIsIgnored)
 	stopPeers(peers)
+	g5.Stop()
 }
 
 func TestEndedGoroutines(t *testing.T) {