[FAB-10970] Make connection refusal more lenient

If the gossip communication layer is called to send a message to a peer
with a given PKI-ID, but after the handshake it discovers the remote peer
has a different PKI-ID than what was expected, it aborts the connection.

This is prolematic for cases where a peer has renewed its certificate, as
the PKI-ID which is a hash on the certificate, won't be the same - and
as a result, the reincarnated peer would be isolated.

This change set makes the connection be aborted only if the peer is
from a different organization.

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

Author: yacovm

gossip/api/channel.go

@@ -18,6 +18,8 @@ func init() {
 	}
 }
 
+//go:generate mockery -dir . -name SecurityAdvisor -case underscore -output ../mocks/
+
 // SecurityAdvisor defines an external auxiliary object
 // that provides security and identity related capabilities
 type SecurityAdvisor interface {

gossip/comm/comm_impl.go

@@ -38,6 +38,13 @@ const (
 	defSendBuffSize = 20
 )
 
+// SecurityAdvisor defines an external auxiliary object
+// that provides security and identity related capabilities
+type SecurityAdvisor interface {
+	// OrgByPeerIdentity returns the organization identity of the given PeerIdentityType
+	OrgByPeerIdentity(api.PeerIdentityType) api.OrgIdentityType
+}
+
 // SetDialTimeout sets the dial timeout
 func SetDialTimeout(timeout time.Duration) {
 	viper.Set("peer.gossip.dialTimeout", timeout)
@@ -53,7 +60,7 @@ func (c *commImpl) SetDialOpts(opts ...grpc.DialOption) {
 
 // NewCommInstanceWithServer creates a comm instance that creates an underlying gRPC server
 func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity api.PeerIdentityType,
-	secureDialOpts api.PeerSecureDialOpts, dialOpts ...grpc.DialOption) (Comm, error) {
+	secureDialOpts api.PeerSecureDialOpts, sa api.SecurityAdvisor, dialOpts ...grpc.DialOption) (Comm, error) {
 
 	var ll net.Listener
 	var s *grpc.Server
@@ -64,6 +71,7 @@ func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity
 	}
 
 	commInst := &commImpl{
+		sa:             sa,
 		pubSub:         util.NewPubSub(),
 		PKIID:          idMapper.GetPKIidOfCert(peerIdentity),
 		idMapper:       idMapper,
@@ -99,10 +107,10 @@ func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity
 
 // NewCommInstance creates a new comm instance that binds itself to the given gRPC server
 func NewCommInstance(s *grpc.Server, certs *common.TLSCertificates, idStore identity.Mapper,
-	peerIdentity api.PeerIdentityType, secureDialOpts api.PeerSecureDialOpts,
+	peerIdentity api.PeerIdentityType, secureDialOpts api.PeerSecureDialOpts, sa api.SecurityAdvisor,
 	dialOpts ...grpc.DialOption) (Comm, error) {
 
-	commInst, err := NewCommInstanceWithServer(-1, idStore, peerIdentity, secureDialOpts, dialOpts...)
+	commInst, err := NewCommInstanceWithServer(-1, idStore, peerIdentity, secureDialOpts, sa, dialOpts...)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -115,6 +123,7 @@ func NewCommInstance(s *grpc.Server, certs *common.TLSCertificates, idStore iden
 }
 
 type commImpl struct {
+	sa             api.SecurityAdvisor
 	tlsCerts       *common.TLSCertificates
 	pubSub         *util.PubSub
 	peerIdentity   api.PeerIdentityType
@@ -175,11 +184,18 @@ func (c *commImpl) createConnection(endpoint string, expectedPKIID common.PKIidT
 		connInfo, err = c.authenticateRemotePeer(stream, true)
 		if err == nil {
 			pkiID = connInfo.ID
+			// PKIID is nil when we don't know the remote PKI id's
 			if expectedPKIID != nil && !bytes.Equal(pkiID, expectedPKIID) {
-				// PKIID is nil when we don't know the remote PKI id's
-				c.logger.Warning("Remote endpoint claims to be a different peer, expected", expectedPKIID, "but got", pkiID)
-				cc.Close()
-				return nil, errors.New("Authentication failure")
+				actualOrg := c.sa.OrgByPeerIdentity(connInfo.Identity)
+				// If the identity isn't present, it's nil - therefore OrgByPeerIdentity would
+				// return nil too and thus would be different than the actual organization
+				identity, _ := c.idMapper.Get(expectedPKIID)
+				oldOrg := c.sa.OrgByPeerIdentity(identity)
+				if !bytes.Equal(actualOrg, oldOrg) {
+					c.logger.Warning("Remote endpoint claims to be a different peer, expected", expectedPKIID, "but got", pkiID)
+					cc.Close()
+					return nil, errors.New("authentication failure")
+				}
 			}
 			conn := newConnection(cl, cc, stream, nil)
 			conn.pkiID = pkiID

gossip/comm/comm_test.go

@@ -25,10 +25,12 @@ import (
 	"github.com/hyperledger/fabric/gossip/api"
 	"github.com/hyperledger/fabric/gossip/common"
 	"github.com/hyperledger/fabric/gossip/identity"
+	"github.com/hyperledger/fabric/gossip/mocks"
 	"github.com/hyperledger/fabric/gossip/util"
 	proto "github.com/hyperledger/fabric/protos/gossip"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -38,6 +40,7 @@ func init() {
 	util.SetupTestLogging()
 	rand.Seed(time.Now().UnixNano())
 	factory.InitFactories(nil)
+	naiveSec.On("OrgByPeerIdentity", mock.Anything).Return(api.OrgIdentityType{})
 }
 
 func acceptAll(msg interface{}) bool {
@@ -54,10 +57,11 @@ var (
 )
 
 type naiveSecProvider struct {
+	mocks.SecurityAdvisor
 }
 
-func (*naiveSecProvider) OrgByPeerIdentity(api.PeerIdentityType) api.OrgIdentityType {
-	return nil
+func (nsp *naiveSecProvider) OrgByPeerIdentity(identity api.PeerIdentityType) api.OrgIdentityType {
+	return nsp.SecurityAdvisor.Called(identity).Get(0).(api.OrgIdentityType)
 }
 
 func (*naiveSecProvider) Expiration(peerIdentity api.PeerIdentityType) (time.Time, error) {
@@ -109,7 +113,7 @@ func (*naiveSecProvider) VerifyByChannel(_ common.ChainID, _ api.PeerIdentityTyp
 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, sec), id, nil)
+	inst, err := NewCommInstanceWithServer(port, identity.NewIdentityMapper(sec, id, noopPurgeIdentity, sec), id, nil, sec)
 	return inst, err
 }
 
@@ -273,7 +277,7 @@ func TestHandshake(t *testing.T) {
 	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()}
-	})
+	}, naiveSec)
 	go s.Serve(ll)
 	assert.NoError(t, err)
 	var msg proto.ReceivedMessage
@@ -396,20 +400,87 @@ func TestBasic(t *testing.T) {
 	waitForMessages(t, out, 2, "Didn't receive 2 messages")
 }
 
+func TestConnectUnexpectedPeer(t *testing.T) {
+	t.Parallel()
+	// Scenarios: In both scenarios, comm1 connects to comm2 or comm3.
+	// and expects to see a PKI-ID which is equal to comm4's PKI-ID.
+	// The connection attempt would succeed or fail based on whether comm2 or comm3
+	// are in the same org as comm4
+	comm1Port := 1548
+	comm2Port := 1549
+	comm3Port := 1550
+	comm4Port := 1551
+
+	identityByPort := func(port int) api.PeerIdentityType {
+		return api.PeerIdentityType(fmt.Sprintf("localhost:%d", port))
+	}
+
+	customNaiveSec := &naiveSecProvider{}
+	customNaiveSec.On("OrgByPeerIdentity", identityByPort(comm1Port)).Return(api.OrgIdentityType("O"))
+	customNaiveSec.On("OrgByPeerIdentity", identityByPort(comm2Port)).Return(api.OrgIdentityType("A"))
+	customNaiveSec.On("OrgByPeerIdentity", identityByPort(comm3Port)).Return(api.OrgIdentityType("B"))
+	customNaiveSec.On("OrgByPeerIdentity", identityByPort(comm4Port)).Return(api.OrgIdentityType("A"))
+
+	comm1, _ := newCommInstance(comm1Port, customNaiveSec)
+	comm2, _ := newCommInstance(comm2Port, naiveSec)
+	comm3, _ := newCommInstance(comm3Port, naiveSec)
+	comm4, _ := newCommInstance(comm4Port, naiveSec)
+
+	defer comm1.Stop()
+	defer comm2.Stop()
+	defer comm3.Stop()
+	defer comm4.Stop()
+
+	messagesForComm1 := comm1.Accept(acceptAll)
+	messagesForComm2 := comm2.Accept(acceptAll)
+	messagesForComm3 := comm3.Accept(acceptAll)
+
+	// Have comm4 send a message to comm1
+	// in order for comm1 to know comm4
+	comm4.Send(createGossipMsg(), remotePeer(comm1Port))
+	<-messagesForComm1
+	// Close the connection with comm4
+	comm1.CloseConn(remotePeer(comm4Port))
+	// At this point, comm1 knows comm4's identity and organization
+
+	t.Run("Same organization", func(t *testing.T) {
+		unexpectedRemotePeer := remotePeer(comm2Port)
+		unexpectedRemotePeer.PKIID = remotePeer(comm4Port).PKIID
+		comm1.Send(createGossipMsg(), unexpectedRemotePeer)
+		select {
+		case <-messagesForComm2:
+		case <-time.After(time.Second * 5):
+			assert.Fail(t, "Didn't receive a message within a timely manner")
+			util.PrintStackTrace()
+		}
+	})
+
+	t.Run("Unexpected organization", func(t *testing.T) {
+		unexpectedRemotePeer := remotePeer(comm3Port)
+		unexpectedRemotePeer.PKIID = remotePeer(comm4Port).PKIID
+		comm1.Send(createGossipMsg(), unexpectedRemotePeer)
+		select {
+		case <-messagesForComm3:
+			assert.Fail(t, "Message shouldn't have been received")
+		case <-time.After(time.Second * 5):
+		}
+	})
+}
+
 func TestProdConstructor(t *testing.T) {
 	t.Parallel()
 	srv, lsnr, dialOpts, certs := createGRPCLayer(20000)
 	defer srv.Stop()
 	defer lsnr.Close()
 	id := []byte("localhost:20000")
-	comm1, _ := NewCommInstance(srv, certs, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity, naiveSec), id, dialOpts)
+	comm1, _ := NewCommInstance(srv, certs, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity, naiveSec), id, dialOpts, naiveSec)
 	go srv.Serve(lsnr)
 
 	srv, lsnr, dialOpts, certs = createGRPCLayer(30000)
 	defer srv.Stop()
 	defer lsnr.Close()
 	id = []byte("localhost:30000")
-	comm2, _ := NewCommInstance(srv, certs, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity, naiveSec), id, dialOpts)
+	comm2, _ := NewCommInstance(srv, certs, identity.NewIdentityMapper(naiveSec, id, noopPurgeIdentity, naiveSec), id, dialOpts, naiveSec)
 	go srv.Serve(lsnr)
 	defer comm1.Stop()
 	defer comm2.Stop()

gossip/gossip/gossip_impl.go

@@ -64,16 +64,16 @@ type gossipServiceImpl struct {
 }
 
 // NewGossipService creates a gossip instance attached to a gRPC server
-func NewGossipService(conf *Config, s *grpc.Server, secAdvisor api.SecurityAdvisor,
+func NewGossipService(conf *Config, s *grpc.Server, sa api.SecurityAdvisor,
 	mcs api.MessageCryptoService, selfIdentity api.PeerIdentityType,
 	secureDialOpts api.PeerSecureDialOpts) Gossip {
 	var err error
 
 	lgr := util.GetLogger(util.LoggingGossipModule, conf.ID)
 
 	g := &gossipServiceImpl{
-		selfOrg:               secAdvisor.OrgByPeerIdentity(selfIdentity),
-		secAdvisor:            secAdvisor,
+		selfOrg:               sa.OrgByPeerIdentity(selfIdentity),
+		secAdvisor:            sa,
 		selfIdentity:          selfIdentity,
 		presumedDead:          make(chan common.PKIidType, presumedDeadChanSize),
 		disc:                  nil,
@@ -91,12 +91,12 @@ 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)
+	}, sa)
 
 	if s == nil {
-		g.comm, err = createCommWithServer(conf.BindPort, g.idMapper, selfIdentity, secureDialOpts)
+		g.comm, err = createCommWithServer(conf.BindPort, g.idMapper, selfIdentity, secureDialOpts, sa)
 	} else {
-		g.comm, err = createCommWithoutServer(s, conf.TLSCerts, g.idMapper, selfIdentity, secureDialOpts)
+		g.comm, err = createCommWithoutServer(s, conf.TLSCerts, g.idMapper, selfIdentity, secureDialOpts, sa)
 	}
 
 	if err != nil {
@@ -159,8 +159,8 @@ func newChannelState(g *gossipServiceImpl) *channelState {
 }
 
 func createCommWithoutServer(s *grpc.Server, certs *common.TLSCertificates, idStore identity.Mapper,
-	identity api.PeerIdentityType, secureDialOpts api.PeerSecureDialOpts) (comm.Comm, error) {
-	return comm.NewCommInstance(s, certs, idStore, identity, secureDialOpts)
+	identity api.PeerIdentityType, secureDialOpts api.PeerSecureDialOpts, sa api.SecurityAdvisor) (comm.Comm, error) {
+	return comm.NewCommInstance(s, certs, idStore, identity, secureDialOpts, sa)
 }
 
 // NewGossipServiceWithServer creates a new gossip instance with a gRPC server
@@ -170,8 +170,8 @@ func NewGossipServiceWithServer(conf *Config, secAdvisor api.SecurityAdvisor, mc
 }
 
 func createCommWithServer(port int, idStore identity.Mapper, identity api.PeerIdentityType,
-	secureDialOpts api.PeerSecureDialOpts) (comm.Comm, error) {
-	return comm.NewCommInstanceWithServer(port, idStore, identity, secureDialOpts)
+	secureDialOpts api.PeerSecureDialOpts, sa api.SecurityAdvisor) (comm.Comm, error) {
+	return comm.NewCommInstanceWithServer(port, idStore, identity, secureDialOpts, sa)
 }
 
 func (g *gossipServiceImpl) toDie() bool {