Skip to content

Commit 5dbe29e

Browse files
yacovmyacovm
committed
[FAB-2198] Introduce envelopes to gossip message
The current gossip implementation uses proto.Marshal and then signs over this output both in the signer and the verifier, which might be dangerous because protobuf marshalling isn't guaranteed to be deterministic. This is the first commit in the series that introduces an envelope message to the GossipMessage, and does some refactoring and preperation for next commit(s) that'll change the stream definition, the signing, and entity-storage related structures. The refactoring basically entails moving the ReceivedMessage from comm to proto, as part of the extensions to the gossip proto types. Signed-off-by: Yacov Manevich <yacovm@il.ibm.com> Change-Id: Ifcedaa87bcf45376a4569ed80b7e7d1b76c87883
1 parent c341fe5 commit 5dbe29e

20 files changed

+232
-176
lines changed

‎gossip/comm/comm.go

Lines changed: 1 addition & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ type Comm interface {
3838

3939
// Accept returns a dedicated read-only channel for messages sent by other nodes that match a certain predicate.
4040
// Each message from the channel can be used to send a reply back to the sender
41-
Accept(common.MessageAcceptor) <-chan ReceivedMessage
41+
Accept(common.MessageAcceptor) <-chan proto.ReceivedMessage
4242

4343
// PresumedDead returns a read-only channel for node endpoints that are suspected to be offline
4444
PresumedDead() <-chan common.PKIidType
@@ -63,20 +63,3 @@ type RemotePeer struct {
6363
func (p *RemotePeer) String() string {
6464
return fmt.Sprintf("%s, PKIid:%v", p.Endpoint, p.PKIID)
6565
}
66-
67-
// ReceivedMessage is a GossipMessage wrapper that
68-
// enables the user to send a message to the origin from which
69-
// the ReceivedMessage was sent from.
70-
// It also allows to know the identity of the sender
71-
type ReceivedMessage interface {
72-
73-
// Respond sends a GossipMessage to the origin from which this ReceivedMessage was sent from
74-
Respond(msg *proto.GossipMessage)
75-
76-
// GetGossipMessage returns the underlying GossipMessage
77-
GetGossipMessage() *proto.GossipMessage
78-
79-
// GetPKIID returns the PKI-ID of the remote peer
80-
// that sent the message
81-
GetPKIID() common.PKIidType
82-
}

‎gossip/comm/comm_impl.go

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,7 @@ func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity
9898
deadEndpoints: make(chan common.PKIidType, 100),
9999
stopping: int32(0),
100100
exitChan: make(chan struct{}, 1),
101-
subscriptions: make([]chan ReceivedMessage, 0),
101+
subscriptions: make([]chan proto.ReceivedMessage, 0),
102102
blackListedPKIIDs: make([]common.PKIidType, 0),
103103
}
104104
commInst.connStore = newConnStore(commInst, commInst.logger)
@@ -154,7 +154,7 @@ type commImpl struct {
154154
exitChan chan struct{}
155155
stopping int32
156156
stopWG sync.WaitGroup
157-
subscriptions []chan ReceivedMessage
157+
subscriptions []chan proto.ReceivedMessage
158158
blackListedPKIIDs []common.PKIidType
159159
}
160160

@@ -290,9 +290,9 @@ func (c *commImpl) Probe(remotePeer *RemotePeer) error {
290290
return err
291291
}
292292

293-
func (c *commImpl) Accept(acceptor common.MessageAcceptor) <-chan ReceivedMessage {
293+
func (c *commImpl) Accept(acceptor common.MessageAcceptor) <-chan proto.ReceivedMessage {
294294
genericChan := c.msgPublisher.AddChannel(acceptor)
295-
specificChan := make(chan ReceivedMessage, 10)
295+
specificChan := make(chan proto.ReceivedMessage, 10)
296296

297297
if c.isStopping() {
298298
c.logger.Warning("Accept() called but comm module is stopping, returning empty channel")

‎gossip/comm/comm_test.go

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ func newCommInstance(port int, sec api.MessageCryptoService) (Comm, error) {
9494
return inst, err
9595
}
9696

97-
func handshaker(endpoint string, comm Comm, t *testing.T, sigMutator func([]byte) []byte, pkiIDmutator func([]byte) []byte) <-chan ReceivedMessage {
97+
func handshaker(endpoint string, comm Comm, t *testing.T, sigMutator func([]byte) []byte, pkiIDmutator func([]byte) []byte) <-chan proto.ReceivedMessage {
9898
c := &commImpl{}
9999
err := generateCertificates("key.pem", "cert.pem")
100100
defer os.Remove("cert.pem")
@@ -189,7 +189,7 @@ func TestBasic(t *testing.T) {
189189
m1 := comm1.Accept(acceptAll)
190190
m2 := comm2.Accept(acceptAll)
191191
out := make(chan uint64, 2)
192-
reader := func(ch <-chan ReceivedMessage) {
192+
reader := func(ch <-chan proto.ReceivedMessage) {
193193
m := <-ch
194194
out <- m.GetGossipMessage().Nonce
195195
}
@@ -228,7 +228,7 @@ func TestBlackListPKIid(t *testing.T) {
228228
defer comm3.Stop()
229229
defer comm4.Stop()
230230

231-
reader := func(instance string, out chan uint64, in <-chan ReceivedMessage) {
231+
reader := func(instance string, out chan uint64, in <-chan proto.ReceivedMessage) {
232232
for {
233233
msg := <-in
234234
if msg == nil {
@@ -328,7 +328,7 @@ func TestResponses(t *testing.T) {
328328
defer comm1.Stop()
329329
defer comm2.Stop()
330330

331-
nonceIncrememter := func(msg ReceivedMessage) ReceivedMessage {
331+
nonceIncrememter := func(msg proto.ReceivedMessage) proto.ReceivedMessage {
332332
msg.GetGossipMessage().Nonce++
333333
return msg
334334
}
@@ -365,11 +365,11 @@ func TestAccept(t *testing.T) {
365365
comm2, _ := newCommInstance(7612, naiveSec)
366366

367367
evenNONCESelector := func(m interface{}) bool {
368-
return m.(ReceivedMessage).GetGossipMessage().Nonce%2 == 0
368+
return m.(proto.ReceivedMessage).GetGossipMessage().Nonce%2 == 0
369369
}
370370

371371
oddNONCESelector := func(m interface{}) bool {
372-
return m.(ReceivedMessage).GetGossipMessage().Nonce%2 != 0
372+
return m.(proto.ReceivedMessage).GetGossipMessage().Nonce%2 != 0
373373
}
374374

375375
evenNONCES := comm1.Accept(evenNONCESelector)
@@ -381,7 +381,7 @@ func TestAccept(t *testing.T) {
381381
out := make(chan uint64, defRecvBuffSize)
382382
sem := make(chan struct{}, 0)
383383

384-
readIntoSlice := func(a *[]uint64, ch <-chan ReceivedMessage) {
384+
readIntoSlice := func(a *[]uint64, ch <-chan proto.ReceivedMessage) {
385385
for m := range ch {
386386
*a = append(*a, m.GetGossipMessage().Nonce)
387387
out <- m.GetGossipMessage().Nonce
@@ -422,7 +422,7 @@ func TestReConnections(t *testing.T) {
422422
comm1, _ := newCommInstance(3611, naiveSec)
423423
comm2, _ := newCommInstance(3612, naiveSec)
424424

425-
reader := func(out chan uint64, in <-chan ReceivedMessage) {
425+
reader := func(out chan uint64, in <-chan proto.ReceivedMessage) {
426426
for {
427427
msg := <-in
428428
if msg == nil {

‎gossip/comm/mock/mock_comm.go

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ type packetMock struct {
4646
type channelMock struct {
4747
accept common.MessageAcceptor
4848

49-
channel chan comm.ReceivedMessage
49+
channel chan proto.ReceivedMessage
5050
}
5151

5252
type commMock struct {
@@ -91,11 +91,19 @@ func (packet *packetMock) Respond(msg *proto.GossipMessage) {
9191
}
9292
}
9393

94+
// GetSourceMessage Returns the SignedGossipMessage the ReceivedMessage was
95+
// constructed with
96+
func (packet *packetMock) GetSourceMessage() *proto.SignedGossipMessage {
97+
return nil
98+
}
99+
94100
// GetGossipMessage returns the underlying GossipMessage
95101
func (packet *packetMock) GetGossipMessage() *proto.GossipMessage {
96102
return packet.msg.(*proto.GossipMessage)
97103
}
98104

105+
// GetPKIID returns the PKI-ID of the remote peer
106+
// that sent the message
99107
func (packet *packetMock) GetPKIID() common.PKIidType {
100108
return nil
101109
}
@@ -151,8 +159,8 @@ func (mock *commMock) Probe(peer *comm.RemotePeer) error {
151159

152160
// Accept returns a dedicated read-only channel for messages sent by other nodes that match a certain predicate.
153161
// Each message from the channel can be used to send a reply back to the sender
154-
func (mock *commMock) Accept(accept common.MessageAcceptor) <-chan comm.ReceivedMessage {
155-
ch := make(chan comm.ReceivedMessage)
162+
func (mock *commMock) Accept(accept common.MessageAcceptor) <-chan proto.ReceivedMessage {
163+
ch := make(chan proto.ReceivedMessage)
156164
mock.acceptors = append(mock.acceptors, &channelMock{accept, ch})
157165
return ch
158166
}

‎gossip/comm/mock/mock_comm_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,8 @@ func TestMockComm(t *testing.T) {
3737
defer comm1.Stop()
3838

3939
msgCh := comm1.Accept(func(message interface{}) bool {
40-
return message.(comm.ReceivedMessage).GetGossipMessage().GetStateRequest() != nil ||
41-
message.(comm.ReceivedMessage).GetGossipMessage().GetStateResponse() != nil
40+
return message.(proto.ReceivedMessage).GetGossipMessage().GetStateRequest() != nil ||
41+
message.(proto.ReceivedMessage).GetGossipMessage().GetStateResponse() != nil
4242
})
4343

4444
comm2 := NewCommMock(second.endpoint, members)

‎gossip/comm/msg.go

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,12 @@ type ReceivedMessageImpl struct {
3030
conn *connection
3131
}
3232

33+
// GetSourceMessage Returns the SignedGossipMessage the ReceivedMessage was
34+
// constructed with
35+
func (m *ReceivedMessageImpl) GetSourceMessage() *proto.SignedGossipMessage {
36+
return nil
37+
}
38+
3339
// Respond sends a msg to the source that sent the ReceivedMessageImpl
3440
func (m *ReceivedMessageImpl) Respond(msg *proto.GossipMessage) {
3541
m.conn.send(msg, func(e error) {})

‎gossip/election/adapter.go

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ import (
2222
"time"
2323

2424
"github.com/hyperledger/fabric/gossip/api"
25-
"github.com/hyperledger/fabric/gossip/comm"
2625
"github.com/hyperledger/fabric/gossip/common"
2726
"github.com/hyperledger/fabric/gossip/discovery"
2827
proto "github.com/hyperledger/fabric/protos/gossip"
@@ -61,7 +60,7 @@ type gossip interface {
6160
// If passThrough is false, the messages are processed by the gossip layer beforehand.
6261
// If passThrough is true, the gossip layer doesn't intervene and the messages
6362
// can be used to send a reply back to the sender
64-
Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan comm.ReceivedMessage)
63+
Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan proto.ReceivedMessage)
6564

6665
// Gossip sends a message to other peers to the network
6766
Gossip(msg *proto.GossipMessage)

‎gossip/election/adapter_test.go

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@ import (
2424
"time"
2525

2626
"github.com/hyperledger/fabric/gossip/api"
27-
"github.com/hyperledger/fabric/gossip/comm"
2827
"github.com/hyperledger/fabric/gossip/common"
2928
"github.com/hyperledger/fabric/gossip/discovery"
3029
proto "github.com/hyperledger/fabric/protos/gossip"
@@ -216,7 +215,7 @@ func (g *peerMockGossip) Peers() []discovery.NetworkMember {
216215
return res
217216
}
218217

219-
func (g *peerMockGossip) Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan comm.ReceivedMessage) {
218+
func (g *peerMockGossip) Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan proto.ReceivedMessage) {
220219
ch := make(chan *proto.GossipMessage, 100)
221220
g.acceptorLock.Lock()
222221
g.acceptors = append(g.acceptors, &mockAcceptor{

‎gossip/gossip/certstore.go

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ import (
2222
"sync"
2323

2424
"github.com/hyperledger/fabric/gossip/api"
25-
"github.com/hyperledger/fabric/gossip/comm"
2625
"github.com/hyperledger/fabric/gossip/common"
2726
"github.com/hyperledger/fabric/gossip/gossip/pull"
2827
"github.com/hyperledger/fabric/gossip/identity"
@@ -63,7 +62,7 @@ func newCertStore(puller pull.Mediator, idMapper identity.Mapper, selfIdentity a
6362

6463
puller.Add(certStore.createIdentityMessage())
6564

66-
puller.RegisterMsgHook(pull.ResponseMsgType, func(_ []string, msgs []*proto.GossipMessage, _ comm.ReceivedMessage) {
65+
puller.RegisterMsgHook(pull.ResponseMsgType, func(_ []string, msgs []*proto.GossipMessage, _ proto.ReceivedMessage) {
6766
for _, msg := range msgs {
6867
pkiID := common.PKIidType(msg.GetPeerIdentity().PkiID)
6968
cert := api.PeerIdentityType(msg.GetPeerIdentity().Cert)
@@ -78,7 +77,7 @@ func newCertStore(puller pull.Mediator, idMapper identity.Mapper, selfIdentity a
7877
return certStore
7978
}
8079

81-
func (cs *certStore) handleMessage(msg comm.ReceivedMessage) {
80+
func (cs *certStore) handleMessage(msg proto.ReceivedMessage) {
8281
if update := msg.GetGossipMessage().GetDataUpdate(); update != nil {
8382
for _, m := range update.Data {
8483
if !m.IsIdentityMsg() {

‎gossip/gossip/certstore_test.go

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,12 @@ type sentMsg struct {
5050
mock.Mock
5151
}
5252

53+
// GetSourceMessage Returns the SignedGossipMessage the ReceivedMessage was
54+
// constructed with
55+
func (s *sentMsg) GetSourceMessage() *proto.SignedGossipMessage {
56+
return nil
57+
}
58+
5359
func (s *sentMsg) Respond(msg *proto.GossipMessage) {
5460
s.Called(msg)
5561
}
@@ -80,30 +86,30 @@ func (m *membershipSvcMock) GetMembership() []discovery.NetworkMember {
8086
}
8187

8288
func TestCertStoreBadSignature(t *testing.T) {
83-
badSignature := func(nonce uint64) comm.ReceivedMessage {
89+
badSignature := func(nonce uint64) proto.ReceivedMessage {
8490
return createUpdateMessage(nonce, createBadlySignedUpdateMessage())
8591
}
8692

8793
testCertificateUpdate(t, badSignature, false)
8894
}
8995

9096
func TestCertStoreMismatchedIdentity(t *testing.T) {
91-
mismatchedIdentity := func(nonce uint64) comm.ReceivedMessage {
97+
mismatchedIdentity := func(nonce uint64) proto.ReceivedMessage {
9298
return createUpdateMessage(nonce, createMismatchedUpdateMessage())
9399
}
94100

95101
testCertificateUpdate(t, mismatchedIdentity, false)
96102
}
97103

98104
func TestCertStoreShouldSucceed(t *testing.T) {
99-
totallyFineIdentity := func(nonce uint64) comm.ReceivedMessage {
105+
totallyFineIdentity := func(nonce uint64) proto.ReceivedMessage {
100106
return createUpdateMessage(nonce, createValidUpdateMessage())
101107
}
102108

103109
testCertificateUpdate(t, totallyFineIdentity, true)
104110
}
105111

106-
func testCertificateUpdate(t *testing.T, updateFactory func(uint64) comm.ReceivedMessage, shouldSucceed bool) {
112+
func testCertificateUpdate(t *testing.T, updateFactory func(uint64) proto.ReceivedMessage, shouldSucceed bool) {
107113
config := pull.PullConfig{
108114
MsgType: proto.PullMsgType_IdentityMsg,
109115
PeerCountToSelect: 1,
@@ -257,7 +263,7 @@ func createValidUpdateMessage() *proto.GossipMessage {
257263
return m
258264
}
259265

260-
func createUpdateMessage(nonce uint64, idMsg *proto.GossipMessage) comm.ReceivedMessage {
266+
func createUpdateMessage(nonce uint64, idMsg *proto.GossipMessage) proto.ReceivedMessage {
261267
update := &proto.GossipMessage{
262268
Tag: proto.GossipMessage_EMPTY,
263269
Content: &proto.GossipMessage_DataUpdate{
@@ -271,7 +277,7 @@ func createUpdateMessage(nonce uint64, idMsg *proto.GossipMessage) comm.Received
271277
return &sentMsg{msg: update}
272278
}
273279

274-
func createDigest(nonce uint64) comm.ReceivedMessage {
280+
func createDigest(nonce uint64) proto.ReceivedMessage {
275281
digest := &proto.GossipMessage{
276282
Tag: proto.GossipMessage_EMPTY,
277283
Content: &proto.GossipMessage_DataDig{

0 commit comments

Comments
 (0)