Skip to content

Commit 0a03e39

Browse files
yacovmyacovm
committed
[FAB-5752] Gossip identity expiration I
This commit: 1) Moves the identity mapper into the gossip package from the service package, since it's not needed in the service package. 2) Extends the constructor of the identity mapper to have a callback that is fired whenever an identity is deleted from the identity mapper. Change-Id: I1c9eaa47c97351518848843a9c4f2e1835f82249 Signed-off-by: yacovm <yacovm@il.ibm.com>
1 parent 297d393 commit 0a03e39

File tree

13 files changed

+56
-63
lines changed

13 files changed

+56
-63
lines changed

‎gossip/comm/comm_test.go

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,10 @@ func acceptAll(msg interface{}) bool {
4444
return true
4545
}
4646

47+
var noopPurgeIdentity = func(_ common.PKIidType, _ api.PeerIdentityType) {
48+
49+
}
50+
4751
var (
4852
naiveSec = &naiveSecProvider{}
4953
hmacKey = []byte{0, 0, 0}
@@ -97,7 +101,7 @@ func (*naiveSecProvider) VerifyByChannel(_ common.ChainID, _ api.PeerIdentityTyp
97101
func newCommInstance(port int, sec api.MessageCryptoService) (Comm, error) {
98102
endpoint := fmt.Sprintf("localhost:%d", port)
99103
id := []byte(endpoint)
100-
inst, err := NewCommInstanceWithServer(port, identity.NewIdentityMapper(sec, id), id, nil)
104+
inst, err := NewCommInstanceWithServer(port, identity.NewIdentityMapper(sec, id, noopPurgeIdentity), id, nil)
101105
return inst, err
102106
}
103107

@@ -215,7 +219,7 @@ func TestHandshake(t *testing.T) {
215219
go s.Serve(ll)
216220

217221
id := []byte("localhost:9611")
218-
idMapper := identity.NewIdentityMapper(naiveSec, id)
222+
idMapper := identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity)
219223
inst, err := NewCommInstance(s, nil, idMapper, api.PeerIdentityType("localhost:9611"), func() []grpc.DialOption {
220224
return []grpc.DialOption{grpc.WithInsecure()}
221225
})
@@ -347,7 +351,7 @@ func TestProdConstructor(t *testing.T) {
347351
defer srv.Stop()
348352
defer lsnr.Close()
349353
id := []byte("localhost:20000")
350-
comm1, _ := NewCommInstance(srv, &peerIdentity, identity.NewIdentityMapper(naiveSec, id), id, dialOpts)
354+
comm1, _ := NewCommInstance(srv, &peerIdentity, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity), id, dialOpts)
351355
comm1.(*commImpl).selfCertHash = certHash
352356
go srv.Serve(lsnr)
353357

@@ -356,7 +360,7 @@ func TestProdConstructor(t *testing.T) {
356360
defer srv.Stop()
357361
defer lsnr.Close()
358362
id = []byte("localhost:30000")
359-
comm2, _ := NewCommInstance(srv, &peerIdentity, identity.NewIdentityMapper(naiveSec, id), id, dialOpts)
363+
comm2, _ := NewCommInstance(srv, &peerIdentity, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity), id, dialOpts)
360364
comm2.(*commImpl).selfCertHash = certHash
361365
go srv.Serve(lsnr)
362366
defer comm1.Stop()

‎gossip/gossip/certstore_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -430,7 +430,7 @@ func createObjects(updateFactory func(uint64) proto.ReceivedMessage, msgCons pro
430430
selfIdentity := api.PeerIdentityType("SELF")
431431
certStore = newCertStore(&pullerMock{
432432
Mediator: pullMediator,
433-
}, identity.NewIdentityMapper(cs, selfIdentity), selfIdentity, cs)
433+
}, identity.NewIdentityMapper(cs, selfIdentity, func(_ common.PKIidType, _ api.PeerIdentityType) {}), selfIdentity, cs)
434434

435435
wg := sync.WaitGroup{}
436436
wg.Add(1)

‎gossip/gossip/gossip_impl.go

Lines changed: 21 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -66,37 +66,24 @@ type gossipServiceImpl struct {
6666
disSecAdap *discoverySecurityAdapter
6767
mcs api.MessageCryptoService
6868
stateInfoMsgStore msgstore.MessageStore
69+
certPuller pull.Mediator
6970
}
7071

7172
// NewGossipService creates a gossip instance attached to a gRPC server
7273
func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvisor,
73-
mcs api.MessageCryptoService, idMapper identity.Mapper, selfIdentity api.PeerIdentityType,
74+
mcs api.MessageCryptoService, selfIdentity api.PeerIdentityType,
7475
secureDialOpts api.PeerSecureDialOpts) Gossip {
75-
76-
var c comm.Comm
7776
var err error
7877

7978
lgr := util.GetLogger(util.LoggingGossipModule, conf.ID)
80-
if s == nil {
81-
c, err = createCommWithServer(conf.BindPort, idMapper, selfIdentity, secureDialOpts)
82-
} else {
83-
c, err = createCommWithoutServer(s, conf.TLSServerCert, idMapper, selfIdentity, secureDialOpts)
84-
}
85-
86-
if err != nil {
87-
lgr.Errorf("Failed instntiating communication layer: %+v", errors.WithStack(err))
88-
return nil
89-
}
9079

9180
g := &gossipServiceImpl{
9281
selfOrg: secAdvisor.OrgByPeerIdentity(selfIdentity),
9382
secAdvisor: secAdvisor,
9483
selfIdentity: selfIdentity,
9584
presumedDead: make(chan common.PKIidType, presumedDeadChanSize),
96-
idMapper: idMapper,
9785
disc: nil,
9886
mcs: mcs,
99-
comm: c,
10087
conf: conf,
10188
ChannelDeMultiplexer: comm.NewChannelDemultiplexer(),
10289
logger: lgr,
@@ -107,6 +94,21 @@ func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvis
10794
}
10895
g.stateInfoMsgStore = g.newStateInfoMsgStore()
10996

97+
g.idMapper = identity.NewIdentityMapper(mcs, selfIdentity, func(pkiID common.PKIidType, identity api.PeerIdentityType) {
98+
g.certPuller.Remove(string(pkiID))
99+
})
100+
101+
if s == nil {
102+
g.comm, err = createCommWithServer(conf.BindPort, g.idMapper, selfIdentity, secureDialOpts)
103+
} else {
104+
g.comm, err = createCommWithoutServer(s, conf.TLSServerCert, g.idMapper, selfIdentity, secureDialOpts)
105+
}
106+
107+
if err != nil {
108+
lgr.Error("Failed instntiating communication layer:", err)
109+
return nil
110+
}
111+
110112
g.chanState = newChannelState(g)
111113
g.emitter = newBatchingEmitter(conf.PropagateIterations,
112114
conf.MaxPropagationBurstSize, conf.MaxPropagationBurstLatency,
@@ -117,7 +119,8 @@ func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvis
117119
g.disc = discovery.NewDiscoveryService(g.selfNetworkMember(), g.discAdapter, g.disSecAdap, g.disclosurePolicy)
118120
g.logger.Info("Creating gossip service with self membership of", g.selfNetworkMember())
119121

120-
g.certStore = newCertStore(g.createCertStorePuller(), idMapper, selfIdentity, mcs)
122+
g.certPuller = g.createCertStorePuller()
123+
g.certStore = newCertStore(g.certPuller, g.idMapper, selfIdentity, mcs)
121124

122125
if g.conf.ExternalEndpoint == "" {
123126
g.logger.Warning("External endpoint is empty, peer will not be accessible outside of its organization")
@@ -168,8 +171,8 @@ func createCommWithoutServer(s *grpc.Server, cert *tls.Certificate, idStore iden
168171

169172
// NewGossipServiceWithServer creates a new gossip instance with a gRPC server
170173
func NewGossipServiceWithServer(conf *Config, secAdvisor api.SecurityAdvisor, mcs api.MessageCryptoService,
171-
mapper identity.Mapper, identity api.PeerIdentityType, secureDialOpts api.PeerSecureDialOpts) Gossip {
172-
return NewGossipService(conf, nil, secAdvisor, mcs, mapper, identity, secureDialOpts)
174+
identity api.PeerIdentityType, secureDialOpts api.PeerSecureDialOpts) Gossip {
175+
return NewGossipService(conf, nil, secAdvisor, mcs, identity, secureDialOpts)
173176
}
174177

175178
func createCommWithServer(port int, idStore identity.Mapper, identity api.PeerIdentityType,

‎gossip/gossip/gossip_test.go

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@ import (
2525
"github.com/hyperledger/fabric/gossip/common"
2626
"github.com/hyperledger/fabric/gossip/discovery"
2727
"github.com/hyperledger/fabric/gossip/gossip/algo"
28-
"github.com/hyperledger/fabric/gossip/identity"
2928
"github.com/hyperledger/fabric/gossip/util"
3029
proto "github.com/hyperledger/fabric/protos/gossip"
3130
"github.com/stretchr/testify/assert"
@@ -227,8 +226,7 @@ func newGossipInstanceWithCustomMCS(portPrefix int, id int, maxMsgCount int, mcs
227226
RequestStateInfoInterval: time.Duration(1) * time.Second,
228227
}
229228
selfId := api.PeerIdentityType(conf.InternalEndpoint)
230-
idMapper := identity.NewIdentityMapper(mcs, selfId)
231-
g := NewGossipServiceWithServer(conf, &orgCryptoService{}, mcs, idMapper,
229+
g := NewGossipServiceWithServer(conf, &orgCryptoService{}, mcs,
232230
selfId, nil)
233231

234232
return g
@@ -260,9 +258,7 @@ func newGossipInstanceWithOnlyPull(portPrefix int, id int, maxMsgCount int, boot
260258

261259
cryptoService := &naiveCryptoService{}
262260
selfId := api.PeerIdentityType(conf.InternalEndpoint)
263-
idMapper := identity.NewIdentityMapper(cryptoService, selfId)
264-
265-
g := NewGossipServiceWithServer(conf, &orgCryptoService{}, cryptoService, idMapper,
261+
g := NewGossipServiceWithServer(conf, &orgCryptoService{}, cryptoService,
266262
selfId, nil)
267263
return g
268264
}

‎gossip/gossip/orgs_test.go

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@ import (
1818
"github.com/hyperledger/fabric/gossip/api"
1919
"github.com/hyperledger/fabric/gossip/common"
2020
"github.com/hyperledger/fabric/gossip/discovery"
21-
"github.com/hyperledger/fabric/gossip/identity"
2221
"github.com/hyperledger/fabric/gossip/util"
2322
proto "github.com/hyperledger/fabric/protos/gossip"
2423
"github.com/stretchr/testify/assert"
@@ -110,8 +109,7 @@ func newGossipInstanceWithExternalEndpoint(portPrefix int, id int, mcs *configur
110109
RequestStateInfoInterval: time.Duration(1) * time.Second,
111110
}
112111
selfId := api.PeerIdentityType(conf.InternalEndpoint)
113-
idMapper := identity.NewIdentityMapper(mcs, selfId)
114-
g := NewGossipServiceWithServer(conf, mcs, mcs, idMapper, selfId,
112+
g := NewGossipServiceWithServer(conf, mcs, mcs, selfId,
115113
nil)
116114

117115
return g

‎gossip/identity/identity.go

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,18 +50,22 @@ type Mapper interface {
5050
ListInvalidIdentities(isSuspected api.PeerSuspector) []common.PKIidType
5151
}
5252

53+
type purgeTrigger func(pkiID common.PKIidType, identity api.PeerIdentityType)
54+
5355
// identityMapperImpl is a struct that implements Mapper
5456
type identityMapperImpl struct {
57+
onPurge purgeTrigger
5558
mcs api.MessageCryptoService
5659
pkiID2Cert map[string]*storedIdentity
5760
sync.RWMutex
5861
selfPKIID string
5962
}
6063

6164
// NewIdentityMapper method, all we need is a reference to a MessageCryptoService
62-
func NewIdentityMapper(mcs api.MessageCryptoService, selfIdentity api.PeerIdentityType) Mapper {
65+
func NewIdentityMapper(mcs api.MessageCryptoService, selfIdentity api.PeerIdentityType, onPurge purgeTrigger) Mapper {
6366
selfPKIID := mcs.GetPKIidOfCert(selfIdentity)
6467
idMapper := &identityMapperImpl{
68+
onPurge: onPurge,
6569
mcs: mcs,
6670
pkiID2Cert: make(map[string]*storedIdentity),
6771
selfPKIID: string(selfPKIID),
@@ -140,6 +144,7 @@ func (is *identityMapperImpl) ListInvalidIdentities(isSuspected api.PeerSuspecto
140144
is.Lock()
141145
defer is.Unlock()
142146
for _, pkiID := range revokedIds {
147+
is.onPurge(pkiID, is.pkiID2Cert[string(pkiID)].peerIdentity)
143148
delete(is.pkiID2Cert, string(pkiID))
144149
}
145150
return revokedIds

‎gossip/identity/identity_test.go

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@ type naiveCryptoService struct {
2929
revokedIdentities map[string]struct{}
3030
}
3131

32+
var noopPurgeTrigger = func(_ common.PKIidType, _ api.PeerIdentityType) {}
33+
3234
func init() {
3335
util.SetupTestLogging()
3436
}
@@ -75,7 +77,7 @@ func (*naiveCryptoService) Verify(peerIdentity api.PeerIdentityType, signature,
7577
}
7678

7779
func TestPut(t *testing.T) {
78-
idStore := NewIdentityMapper(msgCryptoService, dummyID)
80+
idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger)
7981
identity := []byte("yacovm")
8082
identity2 := []byte("not-yacovm")
8183
pkiID := msgCryptoService.GetPKIidOfCert(api.PeerIdentityType(identity))
@@ -88,7 +90,7 @@ func TestPut(t *testing.T) {
8890
}
8991

9092
func TestGet(t *testing.T) {
91-
idStore := NewIdentityMapper(msgCryptoService, dummyID)
93+
idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger)
9294
identity := []byte("yacovm")
9395
identity2 := []byte("not-yacovm")
9496
pkiID := msgCryptoService.GetPKIidOfCert(api.PeerIdentityType(identity))
@@ -103,7 +105,7 @@ func TestGet(t *testing.T) {
103105
}
104106

105107
func TestVerify(t *testing.T) {
106-
idStore := NewIdentityMapper(msgCryptoService, dummyID)
108+
idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger)
107109
identity := []byte("yacovm")
108110
identity2 := []byte("not-yacovm")
109111
pkiID := msgCryptoService.GetPKIidOfCert(api.PeerIdentityType(identity))
@@ -116,7 +118,7 @@ func TestVerify(t *testing.T) {
116118
}
117119

118120
func TestListInvalidIdentities(t *testing.T) {
119-
idStore := NewIdentityMapper(msgCryptoService, dummyID)
121+
idStore := NewIdentityMapper(msgCryptoService, dummyID, noopPurgeTrigger)
120122
identity := []byte("yacovm")
121123
// Test for a revoked identity
122124
pkiID := msgCryptoService.GetPKIidOfCert(api.PeerIdentityType(identity))

‎gossip/integration/integration.go

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@ import (
1515
"github.com/hyperledger/fabric/core/config"
1616
"github.com/hyperledger/fabric/gossip/api"
1717
"github.com/hyperledger/fabric/gossip/gossip"
18-
"github.com/hyperledger/fabric/gossip/identity"
1918
"github.com/hyperledger/fabric/gossip/util"
2019
"github.com/pkg/errors"
2120
"github.com/spf13/viper"
@@ -68,7 +67,7 @@ func newConfig(selfEndpoint string, externalEndpoint string, bootPeers ...string
6867

6968
// NewGossipComponent creates a gossip component that attaches itself to the given gRPC server
7069
func NewGossipComponent(peerIdentity []byte, endpoint string, s *grpc.Server,
71-
secAdv api.SecurityAdvisor, cryptSvc api.MessageCryptoService, idMapper identity.Mapper,
70+
secAdv api.SecurityAdvisor, cryptSvc api.MessageCryptoService,
7271
secureDialOpts api.PeerSecureDialOpts, bootPeers ...string) (gossip.Gossip, error) {
7372

7473
externalEndpoint := viper.GetString("peer.gossip.externalEndpoint")
@@ -77,7 +76,7 @@ func NewGossipComponent(peerIdentity []byte, endpoint string, s *grpc.Server,
7776
if err != nil {
7877
return nil, errors.WithStack(err)
7978
}
80-
gossipInstance := gossip.NewGossipService(conf, s, secAdv, cryptSvc, idMapper,
79+
gossipInstance := gossip.NewGossipService(conf, s, secAdv, cryptSvc,
8180
peerIdentity, secureDialOpts)
8281

8382
return gossipInstance, nil

‎gossip/integration/integration_test.go

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@ import (
1515
"github.com/hyperledger/fabric/core/config"
1616
"github.com/hyperledger/fabric/gossip/api"
1717
"github.com/hyperledger/fabric/gossip/common"
18-
"github.com/hyperledger/fabric/gossip/identity"
1918
"github.com/hyperledger/fabric/gossip/util"
2019
"github.com/hyperledger/fabric/msp/mgmt"
2120
"github.com/hyperledger/fabric/msp/mgmt/testtools"
@@ -52,15 +51,13 @@ func TestNewGossipCryptoService(t *testing.T) {
5251
endpoint3 := "localhost:5613"
5352
msptesttools.LoadMSPSetupForTesting()
5453
peerIdentity, _ := mgmt.GetLocalSigningIdentityOrPanic().Serialize()
55-
idMapper := identity.NewIdentityMapper(cryptSvc, peerIdentity)
56-
57-
g1, err := NewGossipComponent(peerIdentity, endpoint1, s1, secAdv, cryptSvc, idMapper,
54+
g1, err := NewGossipComponent(peerIdentity, endpoint1, s1, secAdv, cryptSvc,
5855
defaultSecureDialOpts)
5956
assert.NoError(t, err)
60-
g2, err := NewGossipComponent(peerIdentity, endpoint2, s2, secAdv, cryptSvc, idMapper,
57+
g2, err := NewGossipComponent(peerIdentity, endpoint2, s2, secAdv, cryptSvc,
6158
defaultSecureDialOpts, endpoint1)
6259
assert.NoError(t, err)
63-
g3, err := NewGossipComponent(peerIdentity, endpoint3, s3, secAdv, cryptSvc, idMapper,
60+
g3, err := NewGossipComponent(peerIdentity, endpoint3, s3, secAdv, cryptSvc,
6461
defaultSecureDialOpts, endpoint1)
6562
assert.NoError(t, err)
6663
defer g1.Stop()
@@ -75,11 +72,10 @@ func TestBadInitialization(t *testing.T) {
7572
msptesttools.LoadMSPSetupForTesting()
7673
peerIdentity, _ := mgmt.GetLocalSigningIdentityOrPanic().Serialize()
7774
s1 := grpc.NewServer()
78-
idMapper := identity.NewIdentityMapper(cryptSvc, peerIdentity)
7975
_, err := newConfig("anEndpointWithoutAPort", "anEndpointWithoutAPort")
8076

8177
viper.Set("peer.tls.enabled", true)
82-
_, err = NewGossipComponent(peerIdentity, "localhost:5000", s1, secAdv, cryptSvc, idMapper,
78+
_, err = NewGossipComponent(peerIdentity, "localhost:5000", s1, secAdv, cryptSvc,
8379
defaultSecureDialOpts)
8480
assert.Error(t, err)
8581
}

‎gossip/service/gossip_service.go

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@ import (
1717
gossipCommon "github.com/hyperledger/fabric/gossip/common"
1818
"github.com/hyperledger/fabric/gossip/election"
1919
"github.com/hyperledger/fabric/gossip/gossip"
20-
"github.com/hyperledger/fabric/gossip/identity"
2120
"github.com/hyperledger/fabric/gossip/integration"
2221
privdata2 "github.com/hyperledger/fabric/gossip/privdata"
2322
"github.com/hyperledger/fabric/gossip/state"
@@ -81,7 +80,6 @@ type gossipServiceImpl struct {
8180
deliveryService deliverclient.DeliverService
8281
deliveryFactory DeliveryServiceFactory
8382
lock sync.RWMutex
84-
idMapper identity.Mapper
8583
mcs api.MessageCryptoService
8684
peerIdentity []byte
8785
secAdv api.SecurityAdvisor
@@ -140,17 +138,15 @@ func InitGossipServiceCustomDeliveryFactory(peerIdentity []byte, endpoint string
140138

141139
logger.Info("Initialize gossip with endpoint", endpoint, "and bootstrap set", bootPeers)
142140

143-
idMapper := identity.NewIdentityMapper(mcs, peerIdentity)
144141
gossip, err = integration.NewGossipComponent(peerIdentity, endpoint, s, secAdv,
145-
mcs, idMapper, secureDialOpts, bootPeers...)
142+
mcs, secureDialOpts, bootPeers...)
146143
gossipServiceInstance = &gossipServiceImpl{
147144
mcs: mcs,
148145
gossipSvc: gossip,
149146
coordinators: make(map[string]privdata2.Coordinator),
150147
chains: make(map[string]state.GossipStateProvider),
151148
leaderElection: make(map[string]election.LeaderElectionService),
152149
deliveryFactory: factory,
153-
idMapper: idMapper,
154150
peerIdentity: peerIdentity,
155151
secAdv: secAdv,
156152
}
@@ -287,7 +283,7 @@ func (g *gossipServiceImpl) Stop() {
287283
}
288284

289285
func (g *gossipServiceImpl) newLeaderElectionComponent(chainID string, callback func(bool)) election.LeaderElectionService {
290-
PKIid := g.idMapper.GetPKIidOfCert(g.peerIdentity)
286+
PKIid := g.mcs.GetPKIidOfCert(g.peerIdentity)
291287
adapter := election.NewAdapter(g, PKIid, gossipCommon.ChainID(chainID))
292288
return election.NewLeaderElectionService(adapter, string(PKIid), callback)
293289
}

0 commit comments

Comments
 (0)