[FAB-4302] Harden gossip bootstrap peer connection

Full details and introduction is in the JIRA item.

Gossip should be protect against mis-configurations that have security
implications, and therefore I suggest to make the bootstrap peer
connection to use the same mechanism as the anchor peer connection -
when a peer connects to an anchor peer, it first performs a handshake
(which is authenticated), and learns about the organization of the peer
behind the endpoint and only then decides whether to send its internal
endpoint or not.

Change-Id: Ieee1abde85f8e22c4ff6adad16dcd6fdab47b57f
Signed-off-by: Yacov Manevich <[email protected]>

Author: yacovm

gossip/discovery/discovery_impl.go

@@ -19,6 +19,7 @@ package discovery
 import (
 	"bytes"
 	"fmt"
+	"math"
 	"strconv"
 	"strings"
 	"sync"
@@ -94,13 +95,14 @@ type gossipDiscoveryImpl struct {
 
 	toDieChan        chan struct{}
 	toDieFlag        int32
+	port             int
 	logger           *logging.Logger
 	disclosurePolicy DisclosurePolicy
 	pubsub           *util.PubSub
 }
 
 // NewDiscoveryService returns a new discovery service with the comm module passed and the crypto service passed
-func NewDiscoveryService(bootstrapPeers []string, self NetworkMember, comm CommService, crypt CryptoService, disPol DisclosurePolicy) Discovery {
+func NewDiscoveryService(self NetworkMember, comm CommService, crypt CryptoService, disPol DisclosurePolicy) Discovery {
 	d := &gossipDiscoveryImpl{
 		self:             self,
 		incTime:          uint64(time.Now().UnixNano()),
@@ -120,6 +122,7 @@ func NewDiscoveryService(bootstrapPeers []string, self NetworkMember, comm CommS
 		pubsub:           util.NewPubSub(),
 	}
 
+	d.validateSelfConfig()
 	d.msgStore = newAliveMsgStore(d)
 
 	go d.periodicalSendAlive()
@@ -128,8 +131,6 @@ func NewDiscoveryService(bootstrapPeers []string, self NetworkMember, comm CommS
 	go d.periodicalReconnectToDead()
 	go d.handlePresumedDeadPeers()
 
-	go d.connect2BootstrapPeers(bootstrapPeers)
-
 	d.logger.Info("Started", self, "incTime is", d.incTime)
 
 	return d
@@ -147,6 +148,13 @@ func (d *gossipDiscoveryImpl) Lookup(PKIID common.PKIidType) *NetworkMember {
 }
 
 func (d *gossipDiscoveryImpl) Connect(member NetworkMember, id identifier) {
+	for _, endpoint := range []string{member.InternalEndpoint, member.Endpoint} {
+		if d.isMyOwnEndpoint(endpoint) {
+			d.logger.Debug("Skipping connecting to myself")
+			return
+		}
+	}
+
 	d.logger.Debug("Entering", member)
 	defer d.logger.Debug("Exiting")
 	go func() {
@@ -175,58 +183,41 @@ func (d *gossipDiscoveryImpl) Connect(member NetworkMember, id identifier) {
 	}()
 }
 
-func (d *gossipDiscoveryImpl) sendUntilAcked(peer *NetworkMember, message *proto.SignedGossipMessage) {
-	nonce := message.Nonce
-	for i := 0; i < maxConnectionAttempts && !d.toDie(); i++ {
-		sub := d.pubsub.Subscribe(fmt.Sprintf("%d", nonce), time.Second*5)
-		d.comm.SendToPeer(peer, message)
-		if _, timeoutErr := sub.Listen(); timeoutErr == nil {
-			return
-		}
-		time.Sleep(getReconnectInterval())
-	}
+func (d *gossipDiscoveryImpl) isMyOwnEndpoint(endpoint string) bool {
+	return endpoint == fmt.Sprintf("127.0.0.1:%d", d.port) || endpoint == fmt.Sprintf("localhost:%d", d.port) ||
+		endpoint == d.self.InternalEndpoint || endpoint == d.self.Endpoint
 }
 
-func (d *gossipDiscoveryImpl) connect2BootstrapPeers(endpoints []string) {
-	if len(d.self.InternalEndpoint) == 0 {
-		d.logger.Panic("Internal endpoint is empty:", d.self.InternalEndpoint)
+func (d *gossipDiscoveryImpl) validateSelfConfig() {
+	endpoint := d.self.InternalEndpoint
+	if len(endpoint) == 0 {
+		d.logger.Panic("Internal endpoint is empty:", endpoint)
 	}
 
-	if len(strings.Split(d.self.InternalEndpoint, ":")) != 2 {
-		d.logger.Panicf("Self endpoint %s isn't formatted as 'host:port'", d.self.InternalEndpoint)
+	internalEndpointSplit := strings.Split(endpoint, ":")
+	if len(internalEndpointSplit) != 2 {
+		d.logger.Panicf("Self endpoint %s isn't formatted as 'host:port'", endpoint)
 	}
-
-	myPort, err := strconv.ParseInt(strings.Split(d.self.InternalEndpoint, ":")[1], 10, 64)
+	myPort, err := strconv.ParseInt(internalEndpointSplit[1], 10, 64)
 	if err != nil {
-		d.logger.Panicf("Self endpoint %s has not valid port'", d.self.InternalEndpoint)
+		d.logger.Panicf("Self endpoint %s has not valid port'", endpoint)
 	}
 
-	d.logger.Info("Entering:", endpoints)
-	defer d.logger.Info("Exiting")
-	endpoints = filterOutLocalhost(endpoints, int(myPort))
-	if len(endpoints) == 0 {
-		return
+	if myPort > int64(math.MaxUint16) {
+		d.logger.Panicf("Self endpoint %s's port takes more than 16 bits", endpoint)
 	}
 
-	for i := 0; i < maxConnectionAttempts && !d.somePeerIsKnown() && !d.toDie(); i++ {
-		var wg sync.WaitGroup
-		req := d.createMembershipRequest(true).NoopSign()
-		wg.Add(len(endpoints))
-		for _, endpoint := range endpoints {
-			go func(endpoint string) {
-				defer wg.Done()
-				peer := &NetworkMember{
-					Endpoint:         endpoint,
-					InternalEndpoint: endpoint,
-				}
-				if !d.comm.Ping(peer) {
-					d.logger.Warning("Failed connecting to bootstrap peer", endpoint)
-					return
-				}
-				d.comm.SendToPeer(peer, req)
-			}(endpoint)
+	d.port = int(myPort)
+}
+
+func (d *gossipDiscoveryImpl) sendUntilAcked(peer *NetworkMember, message *proto.SignedGossipMessage) {
+	nonce := message.Nonce
+	for i := 0; i < maxConnectionAttempts && !d.toDie(); i++ {
+		sub := d.pubsub.Subscribe(fmt.Sprintf("%d", nonce), time.Second*5)
+		d.comm.SendToPeer(peer, message)
+		if _, timeoutErr := sub.Listen(); timeoutErr == nil {
+			return
 		}
-		wg.Wait()
 		time.Sleep(getReconnectInterval())
 	}
 }
@@ -962,17 +953,6 @@ func getReconnectInterval() time.Duration {
 	return util.GetDurationOrDefault("peer.gossip.reconnectInterval", getAliveExpirationTimeout())
 }
 
-func filterOutLocalhost(endpoints []string, port int) []string {
-	var returnedEndpoints []string
-	for _, endpoint := range endpoints {
-		if endpoint == fmt.Sprintf("127.0.0.1:%d", port) || endpoint == fmt.Sprintf("localhost:%d", port) {
-			continue
-		}
-		returnedEndpoints = append(returnedEndpoints, endpoint)
-	}
-	return returnedEndpoints
-}
-
 type aliveMsgStore struct {
 	msgstore.MessageStore
 }

gossip/discovery/discovery_test.go

@@ -314,7 +314,13 @@ func createDiscoveryInstanceThatGossips(port int, id string, bootstrapPeers []st
 	}
 	s := grpc.NewServer()
 
-	discSvc := NewDiscoveryService(bootstrapPeers, self, comm, comm, pol)
+	discSvc := NewDiscoveryService(self, comm, comm, pol)
+	for _, bootPeer := range bootstrapPeers {
+		discSvc.Connect(NetworkMember{Endpoint: bootPeer, InternalEndpoint: bootPeer}, func() (*PeerIdentification, error) {
+			return &PeerIdentification{SelfOrg: true, ID: common.PKIidType(bootPeer)}, nil
+		})
+	}
+
 	gossInst := &gossipInstance{comm: comm, gRGCserv: s, Discovery: discSvc, lsnr: ll, shouldGossip: shouldGossip, port: port}
 
 	proto.RegisterGossipServer(s, gossInst)
@@ -580,7 +586,7 @@ func TestGossipDiscoverySkipConnectingToLocalhostBootstrap(t *testing.T) {
 	inst := createDiscoveryInstance(11611, "d1", []string{"localhost:11611", "127.0.0.1:11611"})
 	inst.comm.lock.Lock()
 	inst.comm.mock = &mock.Mock{}
-	inst.comm.mock.On("SendToPeer", mock.Anything).Run(func(mock.Arguments) {
+	inst.comm.mock.On("SendToPeer", mock.Anything, mock.Anything).Run(func(mock.Arguments) {
 		t.Fatal("Should not have connected to any peer")
 	})
 	inst.comm.mock.On("Ping", mock.Anything).Run(func(mock.Arguments) {
@@ -806,21 +812,6 @@ func TestConfigFromFile(t *testing.T) {
 	assert.Equal(t, time.Duration(25)*time.Second, getReconnectInterval())
 }
 
-func TestFilterOutLocalhost(t *testing.T) {
-	t.Parallel()
-	endpoints := []string{"localhost:5611", "127.0.0.1:5611", "1.2.3.4:5611"}
-	assert.Len(t, filterOutLocalhost(endpoints, 5611), 1)
-	endpoints = []string{"1.2.3.4:5611"}
-	assert.Len(t, filterOutLocalhost(endpoints, 5611), 1)
-	endpoints = []string{"localhost:5611", "127.0.0.1:5611"}
-	assert.Len(t, filterOutLocalhost(endpoints, 5611), 0)
-	// Check slice returned is a copy
-	endpoints = []string{"localhost:5611", "127.0.0.1:5611", "1.2.3.4:5611"}
-	endpoints2 := filterOutLocalhost(endpoints, 5611)
-	endpoints2[0] = "bla bla"
-	assert.NotEqual(t, endpoints[2], endpoints[0])
-}
-
 func TestMsgStoreExpiration(t *testing.T) {
 	// Starts 4 instances, wait for membership to build, stop 2 instances
 	// Check that membership in 2 running instances become 2

gossip/gossip/anchor_test.go

@@ -287,3 +287,69 @@ func TestAnchorPeer(t *testing.T) {
 	pm1.finishedSignal.Wait()
 	pm2.finishedSignal.Wait()
 }
+
+func TestBootstrapPeerMisConfiguration(t *testing.T) {
+	t.Parallel()
+	// Scenario:
+	// The peer 'p' is a peer in orgA
+	// Peers bs1 and bs2 are bootstrap peers.
+	// bs1 is in orgB, so p shouldn't connect to it.
+	// bs2 is in orgA, so p should connect to it.
+	// We test by intercepting *all* messages that bs1 and bs2 get from p, that:
+	// 1) At least 3 connection attempts were sent from p to bs1
+	// 2) A membership request was sent from p to bs2
+
+	cs := &configurableCryptoService{m: make(map[string]api.OrgIdentityType)}
+	portPrefix := 43478
+	orgA := "orgA"
+	orgB := "orgB"
+	cs.putInOrg(portPrefix, orgA)
+	cs.putInOrg(portPrefix+1, orgB)
+	cs.putInOrg(portPrefix+2, orgA)
+
+	onlyHandshakes := func(t *testing.T, index int, m *receivedMsg) {
+		// Ensure all messages sent are connection establishment messages
+		// that are probing attempts
+		assert.NotNil(t, m.GetConn())
+		// If the logic we test in this test- fails,
+		// the first message would be a membership request,
+		// so this assertion would capture it and print a corresponding failure
+		assert.Nil(t, m.GetMemReq())
+	}
+	// Initialize a peer mock that would wait for 3 messages sent to it
+	bs1 := newPeerMock(portPrefix+1, 3, t, onlyHandshakes)
+	defer bs1.stop()
+
+	membershipRequestsSent := make(chan struct{}, 100)
+	detectMembershipRequest := func(t *testing.T, index int, m *receivedMsg) {
+		if m.GetMemReq() != nil {
+			membershipRequestsSent <- struct{}{}
+		}
+	}
+
+	bs2 := newPeerMock(portPrefix+2, 0, t, detectMembershipRequest)
+	defer bs2.stop()
+
+	p := newGossipInstanceWithExternalEndpoint(portPrefix, 0, cs, fmt.Sprintf("localhost:%d", portPrefix), 1, 2)
+	defer p.Stop()
+
+	// Wait for 3 handshake attempts from the bootstrap peer from orgB,
+	// to prove that the peer did try to probe the bootstrap peer from orgB
+	got3Handshakes := make(chan struct{})
+	go func() {
+		bs1.finishedSignal.Wait()
+		got3Handshakes <- struct{}{}
+	}()
+
+	select {
+	case <-got3Handshakes:
+	case <-time.After(time.Second * 15):
+		assert.Fail(t, "Didn't detect 3 handshake attempts to the bootstrap peer from orgB")
+	}
+
+	select {
+	case <-membershipRequestsSent:
+	case <-time.After(time.Second * 15):
+		assert.Fail(t, "Bootstrap peer didn't receive a membership request from the peer within a timely manner")
+	}
+}

gossip/gossip/gossip_impl.go

@@ -124,7 +124,7 @@ func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvis
 
 	g.discAdapter = g.newDiscoveryAdapter()
 	g.disSecAdap = g.newDiscoverySecurityAdapter()
-	g.disc = discovery.NewDiscoveryService(conf.BootstrapPeers, g.selfNetworkMember(), g.discAdapter, g.disSecAdap, g.disclosurePolicy)
+	g.disc = discovery.NewDiscoveryService(g.selfNetworkMember(), g.discAdapter, g.disSecAdap, g.disclosurePolicy)
 	g.logger.Info("Creating gossip service with self membership of", g.selfNetworkMember())
 
 	g.certStore = newCertStore(g.createCertStorePuller(), idMapper, selfIdentity, mcs)
@@ -135,6 +135,7 @@ func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvis
 
 	go g.start()
 	go g.periodicalIdentityValidationAndExpiration()
+	go g.connect2BootstrapPeers()
 
 	return g
 }
@@ -1033,6 +1034,32 @@ func (g *gossipServiceImpl) sameOrgOrOurOrgPullFilter(msg proto.ReceivedMessage)
 	}
 }
 
+func (g *gossipServiceImpl) connect2BootstrapPeers() {
+	for _, endpoint := range g.conf.BootstrapPeers {
+		endpoint := endpoint
+		identifier := func() (*discovery.PeerIdentification, error) {
+			remotePeerIdentity, err := g.comm.Handshake(&comm.RemotePeer{Endpoint: endpoint})
+			if err != nil {
+				return nil, err
+			}
+			sameOrg := bytes.Equal(g.selfOrg, g.secAdvisor.OrgByPeerIdentity(remotePeerIdentity))
+			if !sameOrg {
+				return nil, fmt.Errorf("%s isn't in our organization, cannot be a bootstrap peer", endpoint)
+			}
+			pkiID := g.mcs.GetPKIidOfCert(remotePeerIdentity)
+			if len(pkiID) == 0 {
+				return nil, fmt.Errorf("Wasn't able to extract PKI-ID of remote peer with identity of %v", remotePeerIdentity)
+			}
+			return &discovery.PeerIdentification{ID: pkiID, SelfOrg: sameOrg}, nil
+		}
+		g.disc.Connect(discovery.NetworkMember{
+			InternalEndpoint: endpoint,
+			Endpoint:         endpoint,
+		}, identifier)
+	}
+
+}
+
 func (g *gossipServiceImpl) createStateInfoMsg(metadata []byte, chainID common.ChainID) (*proto.SignedGossipMessage, error) {
 	pkiID := g.comm.GetPKIid()
 	stateInfMsg := &proto.StateInfo{

sampleconfig/core.yaml

@@ -73,8 +73,8 @@ peer:
         # Bootstrap set to initialize gossip with.
         # This is a list of peers that the peer reaches out to at startup.
         # Important: The endpoints here have to be endpoints of peers in the same
-        # organization, because the connection establishment process to these peers
-        # exposes the internal endpoint of this peer.
+        # organization, because the peer would refuse connecting to these endpoints
+        # unless they are in the same organization as the peer.
         bootstrap: 127.0.0.1:7051
 
         # NOTE: orgLeader and useLeaderElection parameters are mutual exclusive