FAB-1046 Gossip identity learning

The current gossip component assumes the upper layers are able to fetch
each remote peer's certificate by its corresponding PKI-ID.
This is true for 0.5 but this isn't the case for v1,
so this commit makes each peer send the other peer its certificate
and each peer creates a mapping between PKIid-->certificate.

The gossip layer now disseminates certificates of peers in the following ways:
1) During the handshake with a remote peer
2) At initialization time, there is a configurable period in which
   the cert is appended to the AliveMessage
3) Peers can now exchange with remote peers certificates by
   comparing the pkiIDs of the certificates. This works with
   the standard built-in pull mechanism.

This commit also moves the comm layer into the gossip, instead
of what was before- where the gossip was given an instance of
the comm.

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

Author: yacovm

gossip/api/channel.go

@@ -27,7 +27,7 @@ import (
 type SecurityAdvisor interface {
 	// IsInMyOrg returns whether the given peer's certificate represents
 	// a peer in the invoker's organization
-	IsInMyOrg(PeerCert) bool
+	IsInMyOrg(PeerIdentityType) bool
 
 	// Verify verifies a JoinChannelMessage, returns nil on success,
 	// and an error on failure
@@ -54,10 +54,7 @@ type JoinChannelMessage interface {
 
 // ChannelMember is a peer's certificate and endpoint (host:port)
 type ChannelMember struct {
-	Cert PeerCert // Cert defines the certificate of the remote peer
-	Host string   // Host is the hostname/ip address of the remote peer
-	Port int      // Port is the port the remote peer is listening on
+	Cert PeerIdentityType // PeerIdentityType defines the certificate of the remote peer
+	Host string           // Host is the hostname/ip address of the remote peer
+	Port int              // Port is the port the remote peer is listening on
 }
-
-// PeerCert defines the cryptographic identity of a peer
-type PeerCert []byte

gossip/api/crypto.go

@@ -39,6 +39,11 @@ type MessageCryptoService interface {
 	// If the verification succeeded, Verify returns nil meaning no error occurred.
 	// If peerCert is nil, then the signature is verified against this peer's verification key.
 	Verify(peerIdentity PeerIdentityType, signature, message []byte) error
+
+	// ValidateIdentity validates the identity of a remote peer.
+	// If the identity is invalid, revoked, expired it returns an error.
+	// Else, returns nil
+	ValidateIdentity(peerIdentity PeerIdentityType) error
 }
 
 // PeerIdentityType is the peer's certificate

gossip/comm/comm.go

@@ -64,23 +64,6 @@ func (p *RemotePeer) String() string {
 	return fmt.Sprintf("%s, PKIid:%v", p.Endpoint, p.PKIID)
 }
 
-// SecurityProvider enables the communication module to perform
-// a handshake that authenticates the client to the server and vice versa
-type SecurityProvider interface {
-
-	// isEnabled returns whether this
-	IsEnabled() bool
-
-	// Sign signs msg with this peers signing key and outputs
-	// the signature if no error occurred.
-	Sign(msg []byte) ([]byte, error)
-
-	// Verify checks that signature if a valid signature of message under vkID's verification key.
-	// If the verification succeeded, Verify returns nil meaning no error occurred.
-	// If vkID is nil, then the signature is verified against this validator's verification key.
-	Verify(vkID, signature, message []byte) error
-}
-
 // ReceivedMessage is a GossipMessage wrapper that
 // enables the user to send a message to the origin from which
 // the ReceivedMessage was sent from

gossip/comm/comm_impl.go

@@ -27,7 +27,9 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/hyperledger/fabric/gossip/api"
 	"github.com/hyperledger/fabric/gossip/common"
+	"github.com/hyperledger/fabric/gossip/identity"
 	"github.com/hyperledger/fabric/gossip/proto"
 	"github.com/hyperledger/fabric/gossip/util"
 	"github.com/op/go-logging"
@@ -62,7 +64,7 @@ func (c *commImpl) SetDialOpts(opts ...grpc.DialOption) {
 }
 
 // NewCommInstanceWithServer creates a comm instance that creates an underlying gRPC server
-func NewCommInstanceWithServer(port int, sec SecurityProvider, pkID common.PKIidType, dialOpts ...grpc.DialOption) (Comm, error) {
+func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity api.PeerIdentityType, dialOpts ...grpc.DialOption) (Comm, error) {
 	var ll net.Listener
 	var s *grpc.Server
 	var secOpt grpc.DialOption
@@ -77,10 +79,11 @@ func NewCommInstanceWithServer(port int, sec SecurityProvider, pkID common.PKIid
 	}
 
 	commInst := &commImpl{
+		PKIID:             idMapper.GetPKIidOfCert(peerIdentity),
+		idMapper:          idMapper,
 		logger:            util.GetLogger(util.LOGGING_COMM_MODULE, fmt.Sprintf("%d", port)),
-		PKIID:             pkID,
+		peerIdentity:      peerIdentity,
 		opts:              dialOpts,
-		sec:               sec,
 		port:              port,
 		lsnr:              ll,
 		gSrv:              s,
@@ -92,15 +95,15 @@ func NewCommInstanceWithServer(port int, sec SecurityProvider, pkID common.PKIid
 		subscriptions:     make([]chan ReceivedMessage, 0),
 		blackListedPKIIDs: make([]common.PKIidType, 0),
 	}
-	commInst.connStore = newConnStore(commInst, pkID, commInst.logger)
+	commInst.connStore = newConnStore(commInst, commInst.logger)
+	commInst.idMapper.Put(idMapper.GetPKIidOfCert(peerIdentity), peerIdentity)
 
 	if port > 0 {
 		go func() {
 			commInst.stopWG.Add(1)
 			defer commInst.stopWG.Done()
 			s.Serve(ll)
 		}()
-
 		proto.RegisterGossipServer(s, commInst)
 	}
 
@@ -110,8 +113,8 @@ func NewCommInstanceWithServer(port int, sec SecurityProvider, pkID common.PKIid
 }
 
 // NewCommInstance creates a new comm instance that binds itself to the given gRPC server
-func NewCommInstance(s *grpc.Server, sec SecurityProvider, PKIID common.PKIidType, dialOpts ...grpc.DialOption) (Comm, error) {
-	commInst, err := NewCommInstanceWithServer(-1, sec, PKIID, dialOpts...)
+func NewCommInstance(s *grpc.Server, idStore identity.Mapper, peerIdentity api.PeerIdentityType, dialOpts ...grpc.DialOption) (Comm, error) {
+	commInst, err := NewCommInstanceWithServer(-1, idStore, peerIdentity, dialOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -120,8 +123,9 @@ func NewCommInstance(s *grpc.Server, sec SecurityProvider, PKIID common.PKIidTyp
 }
 
 type commImpl struct {
+	peerIdentity      api.PeerIdentityType
+	idMapper          identity.Mapper
 	logger            *util.Logger
-	sec               SecurityProvider
 	opts              []grpc.DialOption
 	connStore         *connectionStore
 	PKIID             []byte
@@ -168,7 +172,6 @@ func (c *commImpl) createConnection(endpoint string, expectedPKIID common.PKIidT
 			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, fmt.Errorf("Authentication failure")
 			}
 			conn := newConnection(cl, cc, stream, nil)
@@ -252,18 +255,20 @@ func (c *commImpl) isStopping() bool {
 	return atomic.LoadInt32(&c.stopping) == int32(1)
 }
 
-func (c *commImpl) Probe(peer *RemotePeer) error {
+func (c *commImpl) Probe(remotePeer *RemotePeer) error {
+	endpoint := remotePeer.Endpoint
+	pkiID := remotePeer.PKIID
 	if c.isStopping() {
 		return fmt.Errorf("Stopping")
 	}
-	c.logger.Debug("Entering, endpoint:", peer.Endpoint, "PKIID:", peer.PKIID)
+	c.logger.Debug("Entering, endpoint:", endpoint, "PKIID:", pkiID)
 	var err error
 
 	opts := c.opts
 	if opts == nil {
 		opts = []grpc.DialOption{grpc.WithInsecure(), grpc.WithTimeout(dialTimeout)}
 	}
-	cc, err := grpc.Dial(peer.Endpoint, append(opts, grpc.WithBlock())...)
+	cc, err := grpc.Dial(endpoint, append(opts, grpc.WithBlock())...)
 	if err != nil {
 		c.logger.Debug("Returning", err)
 		return err
@@ -373,45 +378,54 @@ func (c *commImpl) authenticateRemotePeer(stream stream) (common.PKIidType, erro
 	tlsUnique := ExtractTLSUnique(ctx)
 	var sig []byte
 	var err error
-	if tlsUnique != nil && c.sec.IsEnabled() {
-		sig, err = c.sec.Sign(tlsUnique)
+	if tlsUnique != nil {
+		sig, err = c.idMapper.Sign(tlsUnique)
 		if err != nil {
 			c.logger.Error("Failed signing TLS-Unique:", err)
 			return nil, err
 		}
 	}
 
-	cMsg := createConnectionMsg(c.PKIID, sig)
+	cMsg := createConnectionMsg(c.PKIID, sig, c.peerIdentity)
+	c.logger.Debug("Sending", cMsg, "to", remoteAddress)
 	stream.Send(cMsg)
 	m := readWithTimeout(stream, defConnTimeout)
 	if m == nil {
 		c.logger.Warning("Timed out waiting for connection message from", remoteAddress)
 		return nil, fmt.Errorf("Timed out")
 	}
-	connMsg := m.GetConn()
-	if connMsg == nil {
-		c.logger.Warning("Expected connection message but got", connMsg)
+	receivedMsg := m.GetConn()
+	if receivedMsg == nil {
+		c.logger.Warning("Expected connection message but got", receivedMsg)
 		return nil, fmt.Errorf("Wrong type")
 	}
-	if c.isPKIblackListed(connMsg.PkiID) {
+
+	if receivedMsg.PkiID == nil {
+		c.logger.Warning("%s didn't send a pkiID")
+		return nil, fmt.Errorf("%s didn't send a pkiID", remoteAddress)
+	}
+
+	if c.isPKIblackListed(receivedMsg.PkiID) {
 		c.logger.Warning("Connection attempt from", remoteAddress, "but it is black-listed")
 		return nil, fmt.Errorf("Black-listed")
 	}
+	c.logger.Debug("Received", receivedMsg, "from", remoteAddress)
+	err = c.idMapper.Put(receivedMsg.PkiID, receivedMsg.Cert)
+	if err != nil {
+		c.logger.Warning("Identity store rejected", remoteAddress, ":", err)
+		return nil, err
+	}
 
-	if tlsUnique != nil && c.sec.IsEnabled() {
-		err = c.sec.Verify(connMsg.PkiID, connMsg.Sig, tlsUnique)
+	if tlsUnique != nil {
+		err = c.idMapper.Verify(receivedMsg.PkiID, receivedMsg.Sig, tlsUnique)
 		if err != nil {
 			c.logger.Error("Failed verifying signature from", remoteAddress, ":", err)
 			return nil, err
 		}
 	}
 
-	if connMsg.PkiID == nil {
-		return nil, fmt.Errorf("%s didn't send a pkiID", "Didn't send a pkiID")
-	}
-
 	c.logger.Debug("Authenticated", remoteAddress)
-	return connMsg.PkiID, nil
+	return receivedMsg.PkiID, nil
 
 }
 
@@ -486,12 +500,13 @@ func readWithTimeout(stream interface{}, timeout time.Duration) *proto.GossipMes
 	}
 }
 
-func createConnectionMsg(pkiID common.PKIidType, sig []byte) *proto.GossipMessage {
+func createConnectionMsg(pkiID common.PKIidType, sig []byte, cert api.PeerIdentityType) *proto.GossipMessage {
 	return &proto.GossipMessage{
 		Tag:   proto.GossipMessage_EMPTY,
 		Nonce: 0,
 		Content: &proto.GossipMessage_Conn{
 			Conn: &proto.ConnEstablish{
+				Cert:  cert,
 				PkiID: pkiID,
 				Sig:   sig,
 			},