[FAB-7552] IdentityInfo in gossip identityStore

The discovery service needs to output identities of known peers.

Gossip stores the identities in an identity mapper,
which could be extended to return a snapshot of its identities.

This change set makes it so that the identity mapper also saves
the organizations of peers upon their insertion, and adds
the needed method.

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

Author: yacovm

gossip/comm/comm_test.go

@@ -56,6 +56,10 @@ var (
 type naiveSecProvider struct {
 }
 
+func (*naiveSecProvider) OrgByPeerIdentity(api.PeerIdentityType) api.OrgIdentityType {
+	return nil
+}
+
 func (*naiveSecProvider) Expiration(peerIdentity api.PeerIdentityType) (time.Time, error) {
 	return time.Now().Add(time.Hour), nil
 }
@@ -102,10 +106,10 @@ func (*naiveSecProvider) VerifyByChannel(_ common.ChainID, _ api.PeerIdentityTyp
 	return nil
 }
 
-func newCommInstance(port int, sec api.MessageCryptoService) (Comm, error) {
+func newCommInstance(port int, sec *naiveSecProvider) (Comm, error) {
 	endpoint := fmt.Sprintf("localhost:%d", port)
 	id := []byte(endpoint)
-	inst, err := NewCommInstanceWithServer(port, identity.NewIdentityMapper(sec, id, noopPurgeIdentity), id, nil)
+	inst, err := NewCommInstanceWithServer(port, identity.NewIdentityMapper(sec, id, noopPurgeIdentity, sec), id, nil)
 	return inst, err
 }
 
@@ -222,7 +226,7 @@ func TestHandshake(t *testing.T) {
 	assert.NoError(t, err)
 	s := grpc.NewServer()
 	id := []byte("localhost:9611")
-	idMapper := identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity)
+	idMapper := identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity, naiveSec)
 	inst, err := NewCommInstance(s, nil, idMapper, api.PeerIdentityType("localhost:9611"), func() []grpc.DialOption {
 		return []grpc.DialOption{grpc.WithInsecure()}
 	})
@@ -354,14 +358,14 @@ func TestProdConstructor(t *testing.T) {
 	defer srv.Stop()
 	defer lsnr.Close()
 	id := []byte("localhost:29000")
-	comm1, _ := NewCommInstance(srv, certs, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity), id, dialOpts)
+	comm1, _ := NewCommInstance(srv, certs, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity, naiveSec), id, dialOpts)
 	go srv.Serve(lsnr)
 
 	srv, lsnr, dialOpts, certs = createGRPCLayer(39000)
 	defer srv.Stop()
 	defer lsnr.Close()
 	id = []byte("localhost:39000")
-	comm2, _ := NewCommInstance(srv, certs, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity), id, dialOpts)
+	comm2, _ := NewCommInstance(srv, certs, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity, naiveSec), id, dialOpts)
 	go srv.Serve(lsnr)
 	defer comm1.Stop()
 	defer comm2.Stop()

gossip/gossip/certstore_test.go

@@ -437,7 +437,7 @@ func createObjects(updateFactory func(uint64) proto.ReceivedMessage, msgCons pro
 		Mediator: pullMediator,
 	}, identity.NewIdentityMapper(cs, selfIdentity, func(pkiID common.PKIidType, _ api.PeerIdentityType) {
 		pullMediator.Remove(string(pkiID))
-	}), selfIdentity, cs)
+	}, cs), selfIdentity, cs)
 
 	wg := sync.WaitGroup{}
 	wg.Add(1)

gossip/gossip/gossip.go

@@ -72,6 +72,9 @@ type Gossip interface {
 	// any connections to peers with identities that are found invalid
 	SuspectPeers(s api.PeerSuspector)
 
+	// IdentityInfo returns information known peer identities
+	IdentityInfo() api.PeerIdentitySet
+
 	// Stop stops the gossip component
 	Stop()
 }

gossip/gossip/gossip_impl.go

@@ -91,7 +91,7 @@ func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvis
 	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))
-	})
+	}, secAdvisor)
 
 	if s == nil {
 		g.comm, err = createCommWithServer(conf.BindPort, g.idMapper, selfIdentity, secureDialOpts)
@@ -604,6 +604,11 @@ func (g *gossipServiceImpl) removeSelfLoop(msg *emittedGossipMessage, peers []*c
 	return result
 }
 
+// IdentityInfo returns information known peer identities
+func (g *gossipServiceImpl) IdentityInfo() api.PeerIdentitySet {
+	return g.idMapper.IdentityInfo()
+}
+
 // SendByCriteria sends a given message to all peers that match the given SendCriteria
 func (g *gossipServiceImpl) SendByCriteria(msg *proto.SignedGossipMessage, criteria SendCriteria) error {
 	if criteria.Timeout == 0 {

gossip/gossip/gossip_test.go

@@ -126,6 +126,10 @@ type naiveCryptoService struct {
 	revokedPkiIDS map[string]struct{}
 }
 
+func (cs *naiveCryptoService) OrgByPeerIdentity(api.PeerIdentityType) api.OrgIdentityType {
+	return nil
+}
+
 func (*naiveCryptoService) Expiration(peerIdentity api.PeerIdentityType) (time.Time, error) {
 	if exp, exists := expirationTimes[string(peerIdentity)]; exists {
 		return exp, nil

gossip/identity/identity.go

@@ -47,6 +47,9 @@ type Mapper interface {
 	// SuspectPeers re-validates all peers that match the given predicate
 	SuspectPeers(isSuspected api.PeerSuspector)
 
+	// IdentityInfo returns information known peer identities
+	IdentityInfo() api.PeerIdentitySet
+
 	// Stop stops all background computations of the Mapper
 	Stop()
 }
@@ -57,6 +60,7 @@ type purgeTrigger func(pkiID common.PKIidType, identity api.PeerIdentityType)
 type identityMapperImpl struct {
 	onPurge    purgeTrigger
 	mcs        api.MessageCryptoService
+	sa         api.SecurityAdvisor
 	pkiID2Cert map[string]*storedIdentity
 	sync.RWMutex
 	stopChan chan struct{}
@@ -65,14 +69,15 @@ type identityMapperImpl struct {
 }
 
 // NewIdentityMapper method, all we need is a reference to a MessageCryptoService
-func NewIdentityMapper(mcs api.MessageCryptoService, selfIdentity api.PeerIdentityType, onPurge purgeTrigger) Mapper {
+func NewIdentityMapper(mcs api.MessageCryptoService, selfIdentity api.PeerIdentityType, onPurge purgeTrigger, sa api.SecurityAdvisor) Mapper {
 	selfPKIID := mcs.GetPKIidOfCert(selfIdentity)
 	idMapper := &identityMapperImpl{
 		onPurge:    onPurge,
 		mcs:        mcs,
 		pkiID2Cert: make(map[string]*storedIdentity),
 		stopChan:   make(chan struct{}),
 		selfPKIID:  string(selfPKIID),
+		sa:         sa,
 	}
 	if err := idMapper.Put(selfPKIID, selfIdentity); err != nil {
 		panic(errors.Wrap(err, "Failed putting our own identity into the identity mapper"))
@@ -138,7 +143,7 @@ func (is *identityMapperImpl) Put(pkiID common.PKIidType, identity api.PeerIdent
 		})
 	}
 
-	is.pkiID2Cert[string(id)] = newStoredIdentity(pkiID, identity, expirationTimer)
+	is.pkiID2Cert[string(id)] = newStoredIdentity(pkiID, identity, expirationTimer, is.sa.OrgByPeerIdentity(identity))
 	return nil
 }
 
@@ -210,6 +215,21 @@ func (is *identityMapperImpl) validateIdentities(isSuspected api.PeerSuspector)
 	return revokedIdentities
 }
 
+// IdentityInfo returns information known peer identities
+func (is *identityMapperImpl) IdentityInfo() api.PeerIdentitySet {
+	var res api.PeerIdentitySet
+	is.RLock()
+	defer is.RUnlock()
+	for _, storedIdentity := range is.pkiID2Cert {
+		res = append(res, api.PeerIdentityInfo{
+			Identity:     storedIdentity.peerIdentity,
+			PKIId:        storedIdentity.pkiID,
+			Organization: storedIdentity.orgId,
+		})
+	}
+	return res
+}
+
 func (is *identityMapperImpl) delete(pkiID common.PKIidType, identity api.PeerIdentityType) {
 	is.Lock()
 	defer is.Unlock()
@@ -221,15 +241,17 @@ type storedIdentity struct {
 	pkiID           common.PKIidType
 	lastAccessTime  int64
 	peerIdentity    api.PeerIdentityType
+	orgId           api.OrgIdentityType
 	expirationTimer *time.Timer
 }
 
-func newStoredIdentity(pkiID common.PKIidType, identity api.PeerIdentityType, expirationTimer *time.Timer) *storedIdentity {
+func newStoredIdentity(pkiID common.PKIidType, identity api.PeerIdentityType, expirationTimer *time.Timer, org api.OrgIdentityType) *storedIdentity {
 	return &storedIdentity{
 		pkiID:           pkiID,
 		lastAccessTime:  time.Now().UnixNano(),
 		peerIdentity:    identity,
 		expirationTimer: expirationTimer,
+		orgId:           org,
 	}
 }
 

gossip/identity/identity_test.go

@@ -13,6 +13,8 @@ import (
 	"testing"
 	"time"
 
+	"strings"
+
 	"github.com/hyperledger/fabric/gossip/api"
 	"github.com/hyperledger/fabric/gossip/common"
 	"github.com/hyperledger/fabric/gossip/util"
@@ -40,6 +42,19 @@ func init() {
 	msgCryptoService.On("Expiration", api.PeerIdentityType("invalidIdentity")).Return(time.Now().Add(time.Hour), nil)
 }
 
+func (cs *naiveCryptoService) OrgByPeerIdentity(id api.PeerIdentityType) api.OrgIdentityType {
+	found := false
+	for _, call := range cs.Mock.ExpectedCalls {
+		if call.Method == "OrgByPeerIdentity" {
+			found = true
+		}
+	}
+	if !found {
+		return nil
+	}
+	return cs.Called(id).Get(0).(api.OrgIdentityType)
+}
+
 func (cs *naiveCryptoService) Expiration(peerIdentity api.PeerIdentityType) (time.Time, error) {
 	args := cs.Called(peerIdentity)
 	t, err := args.Get(0), args.Get(1)
@@ -91,7 +106,7 @@ func (*naiveCryptoService) Verify(peerIdentity api.PeerIdentityType, signature,
 }
 
 func TestPut(t *testing.T) {
-	idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger)
+	idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger, msgCryptoService)
 	identity := []byte("yacovm")
 	identity2 := []byte("not-yacovm")
 	identity3 := []byte("invalidIdentity")
@@ -109,7 +124,7 @@ func TestPut(t *testing.T) {
 }
 
 func TestGet(t *testing.T) {
-	idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger)
+	idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger, msgCryptoService)
 	identity := []byte("yacovm")
 	identity2 := []byte("not-yacovm")
 	pkiID := msgCryptoService.GetPKIidOfCert(api.PeerIdentityType(identity))
@@ -124,7 +139,7 @@ func TestGet(t *testing.T) {
 }
 
 func TestVerify(t *testing.T) {
-	idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger)
+	idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger, msgCryptoService)
 	identity := []byte("yacovm")
 	identity2 := []byte("not-yacovm")
 	pkiID := msgCryptoService.GetPKIidOfCert(api.PeerIdentityType(identity))
@@ -152,7 +167,7 @@ func TestListInvalidIdentities(t *testing.T) {
 	selfPKIID := msgCryptoService.GetPKIidOfCert(dummyID)
 	idStore := NewIdentityMapper(msgCryptoService, dummyID, func(_ common.PKIidType, identity api.PeerIdentityType) {
 		deletedIdentities <- string(identity)
-	})
+	}, msgCryptoService)
 	identity := []byte("yacovm")
 	// Test for a revoked identity
 	pkiID := msgCryptoService.GetPKIidOfCert(api.PeerIdentityType(identity))
@@ -228,7 +243,7 @@ func TestExpiration(t *testing.T) {
 	SetIdentityUsageThreshold(time.Second * 500)
 	idStore := NewIdentityMapper(msgCryptoService, dummyID, func(_ common.PKIidType, identity api.PeerIdentityType) {
 		deletedIdentities <- string(identity)
-	})
+	}, msgCryptoService)
 	assertDeletedIdentity := func(expected string) {
 		select {
 		case <-time.After(time.Second * 10):
@@ -296,6 +311,29 @@ func TestExpirationPanic(t *testing.T) {
 	identity3 := []byte("invalidIdentity")
 	msgCryptoService.revokedIdentities[string(identity3)] = struct{}{}
 	assert.Panics(t, func() {
-		NewIdentityMapper(msgCryptoService, identity3, noopPurgeTrigger)
+		NewIdentityMapper(msgCryptoService, identity3, noopPurgeTrigger, msgCryptoService)
 	})
 }
+
+func TestIdentityInfo(t *testing.T) {
+	cs := &naiveCryptoService{}
+	alice := api.PeerIdentityType("alicePeer")
+	bob := api.PeerIdentityType("bobPeer")
+	aliceID := cs.GetPKIidOfCert(alice)
+	bobId := cs.GetPKIidOfCert(bob)
+	cs.On("OrgByPeerIdentity", dummyID).Return(api.OrgIdentityType("D"))
+	cs.On("OrgByPeerIdentity", alice).Return(api.OrgIdentityType("A"))
+	cs.On("OrgByPeerIdentity", bob).Return(api.OrgIdentityType("B"))
+	cs.On("Expiration", mock.Anything).Return(time.Now().Add(time.Minute), nil)
+	idStore := NewIdentityMapper(cs, dummyID, noopPurgeTrigger, cs)
+	idStore.Put(aliceID, alice)
+	idStore.Put(bobId, bob)
+	for org, id := range idStore.IdentityInfo().ByOrg() {
+		identity := string(id[0].Identity)
+		pkiID := string(id[0].PKIId)
+		orgId := string(id[0].Organization)
+		assert.Equal(t, org, orgId)
+		assert.Equal(t, strings.ToLower(org), string(identity[0]))
+		assert.Equal(t, strings.ToLower(org), string(pkiID[0]))
+	}
+}

gossip/service/join_test.go

@@ -92,6 +92,10 @@ func (g *gossipMock) LeaveChan(chainID common.ChainID) {
 	panic("implement me")
 }
 
+func (g *gossipMock) IdentityInfo() api.PeerIdentitySet {
+	panic("implement me")
+}
+
 func (*gossipMock) Stop() {
 	panic("implement me")
 }

gossip/state/mocks/gossip.go

@@ -77,6 +77,11 @@ func (g *GossipMock) Accept(acceptor common.MessageAcceptor, passThrough bool) (
 func (g *GossipMock) JoinChan(joinMsg api.JoinChannelMessage, chainID common.ChainID) {
 }
 
+// IdentityInfo returns information known peer identities
+func (g *GossipMock) IdentityInfo() api.PeerIdentitySet {
+	panic("not implemented")
+}
+
 func (g *GossipMock) Stop() {
 
 }