[FAB-3109] Fix anchor peer connection logic

The current implementation of what happens when the peer connects
to an anchor peer has a flawed logic.

It assumes that if the probe is successful, the sending is also.
That's too optimistic.
Another problem is that after the handshake is performed, the remote peer
may not distinguish between the closing of the connection of the handshake
and the closing of the new connection that sends the MembershipRequest,
since we only support 1 connection between each peer (to prevent DOS).

I re-implemented the logic to do the following:
1) First, perform a handshake to know whether the remote peer is from
  our org or not.
  Peers in different orgs should not know of our internal endpoints).
  The handshake is performed until it succeeds or too many attempts
  are made.
2) Now, construct a MembershipRequest and send it until you get
   acked from the remote peer, or when attempts are exhausted.

Added a test that simulates the flow.

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

Author: yacovm

gossip/comm/comm_impl.go

@@ -626,7 +626,7 @@ func createGRPCLayer(port int) (*grpc.Server, net.Listener, api.PeerSecureDialOp
 	defer os.Remove(keyFileName)
 	defer os.Remove(certFileName)
 
-	err = generateCertificates(keyFileName, certFileName)
+	err = GenerateCertificates(keyFileName, certFileName)
 	if err == nil {
 		cert, err := tls.LoadX509KeyPair(certFileName, keyFileName)
 		if err != nil {

gossip/comm/comm_test.go

@@ -111,7 +111,7 @@ func newCommInstance(port int, sec api.MessageCryptoService) (Comm, error) {
 
 func handshaker(endpoint string, comm Comm, t *testing.T, sigMutator func([]byte) []byte, pkiIDmutator func([]byte) []byte, mutualTLS bool) <-chan proto.ReceivedMessage {
 	c := &commImpl{}
-	err := generateCertificates("key.pem", "cert.pem")
+	err := GenerateCertificates("key.pem", "cert.pem")
 	assert.NoError(t, err, "%v", err)
 	defer os.Remove("cert.pem")
 	defer os.Remove("key.pem")
@@ -289,7 +289,7 @@ func TestProdConstructor(t *testing.T) {
 	keyFileName := fmt.Sprintf("key.%d.pem", util.RandomUInt64())
 	certFileName := fmt.Sprintf("cert.%d.pem", util.RandomUInt64())
 
-	generateCertificates(keyFileName, certFileName)
+	GenerateCertificates(keyFileName, certFileName)
 	cert, _ := tls.LoadX509KeyPair(certFileName, keyFileName)
 	os.Remove(keyFileName)
 	os.Remove(certFileName)
@@ -300,7 +300,7 @@ func TestProdConstructor(t *testing.T) {
 	comm1.(*commImpl).selfCertHash = certHash
 	go srv.Serve(lsnr)
 
-	generateCertificates(keyFileName, certFileName)
+	GenerateCertificates(keyFileName, certFileName)
 	cert, _ = tls.LoadX509KeyPair(certFileName, keyFileName)
 	os.Remove(keyFileName)
 	os.Remove(certFileName)
@@ -350,7 +350,7 @@ func TestCloseConn(t *testing.T) {
 	defer comm1.Stop()
 	acceptChan := comm1.Accept(acceptAll)
 
-	err := generateCertificates("key.pem", "cert.pem")
+	err := GenerateCertificates("key.pem", "cert.pem")
 	assert.NoError(t, err, "%v", err)
 	defer os.Remove("cert.pem")
 	defer os.Remove("key.pem")

gossip/comm/crypto.go

@@ -43,7 +43,7 @@ func writeFile(filename string, keyType string, data []byte) error {
 	return pem.Encode(f, &pem.Block{Type: keyType, Bytes: data})
 }
 
-func generateCertificates(privKeyFile string, certKeyFile string) error {
+func GenerateCertificates(privKeyFile string, certKeyFile string) error {
 	privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
 	if err != nil {
 		return err

gossip/comm/crypto_test.go

@@ -84,7 +84,7 @@ func (s *gossipTestServer) Ping(context.Context, *proto.Empty) (*proto.Empty, er
 }
 
 func TestCertificateExtraction(t *testing.T) {
-	err := generateCertificates("key.pem", "cert.pem")
+	err := GenerateCertificates("key.pem", "cert.pem")
 	defer os.Remove("cert.pem")
 	defer os.Remove("key.pem")
 	assert.NoError(t, err, "%v", err)
@@ -94,7 +94,7 @@ func TestCertificateExtraction(t *testing.T) {
 	srv := createTestServer(t, &serverCert)
 	defer srv.stop()
 
-	generateCertificates("key2.pem", "cert2.pem")
+	GenerateCertificates("key2.pem", "cert2.pem")
 	defer os.Remove("cert2.pem")
 	defer os.Remove("key2.pem")
 	clientCert, err := tls.LoadX509KeyPair("cert2.pem", "key2.pem")

gossip/discovery/discovery.go

@@ -97,6 +97,16 @@ func (n NetworkMember) PreferredEndpoint() string {
 	return n.Endpoint
 }
 
+// PeerIdentification encompasses a remote peer's
+// PKI-ID and whether its in the same org as the current
+// peer or not
+type PeerIdentification struct {
+	ID      common.PKIidType
+	SelfOrg bool
+}
+
+type identifier func() (*PeerIdentification, error)
+
 // Discovery is the interface that represents a discovery module
 type Discovery interface {
 
@@ -123,7 +133,8 @@ type Discovery interface {
 	InitiateSync(peerNum int)
 
 	// Connect makes this instance to connect to a remote instance
-	// The sendInternalEndpoint param determines whether or not
-	// to include the internal endpoint in the membership request,
-	Connect(member NetworkMember, sendInternalEndpoint func() bool)
+	// The identifier param is a function that can be used to identify
+	// the peer, and to assert its PKI-ID, whether its in the peer's org or not,
+	// and whether the action was successful or not
+	Connect(member NetworkMember, id identifier)
 }

gossip/discovery/discovery_impl.go

@@ -60,6 +60,12 @@ func SetReconnectInterval(interval time.Duration) {
 	viper.Set("peer.gossip.reconnectInterval", interval)
 }
 
+// SetMaxConnAttempts sets the maximum number of connection
+// attempts the peer would perform when invoking Connect()
+func SetMaxConnAttempts(attempts int) {
+	maxConnectionAttempts = attempts
+}
+
 type timestamp struct {
 	incTime  time.Time
 	seqNum   uint64
@@ -92,6 +98,7 @@ type gossipDiscoveryImpl struct {
 	toDieFlag        int32
 	logger           *logging.Logger
 	disclosurePolicy DisclosurePolicy
+	pubsub           *util.PubSub
 }
 
 // NewDiscoveryService returns a new discovery service with the comm module passed and the crypto service passed
@@ -112,6 +119,7 @@ func NewDiscoveryService(bootstrapPeers []string, self NetworkMember, comm CommS
 		toDieFlag:        int32(0),
 		logger:           util.GetLogger(util.LoggingDiscoveryModule, self.InternalEndpoint),
 		disclosurePolicy: disPol,
+		pubsub:           util.NewPubSub(),
 	}
 
 	d.msgStore = newAliveMsgStore(d)
@@ -137,32 +145,47 @@ func (d *gossipDiscoveryImpl) Lookup(PKIID common.PKIidType) *NetworkMember {
 	return nm
 }
 
-func (d *gossipDiscoveryImpl) Connect(member NetworkMember, sendInternalEndpoint func() bool) {
+func (d *gossipDiscoveryImpl) Connect(member NetworkMember, id identifier) {
 	d.logger.Debug("Entering", member)
 	defer d.logger.Debug("Exiting")
-
 	go func() {
 		for i := 0; i < maxConnectionAttempts && !d.toDie(); i++ {
-			peer := &NetworkMember{
-				InternalEndpoint: member.InternalEndpoint,
-				Endpoint:         member.Endpoint,
-			}
-
-			if !d.comm.Ping(peer) {
+			id, err := id()
+			if err != nil {
 				if d.toDie() {
 					return
 				}
-				d.logger.Warning("Could not connect to", member)
+				d.logger.Warning("Could not connect to", member, ":", err)
 				time.Sleep(getReconnectInterval())
 				continue
 			}
-			req := d.createMembershipRequest(sendInternalEndpoint()).NoopSign()
-			d.comm.SendToPeer(peer, req)
+			peer := &NetworkMember{
+				InternalEndpoint: member.InternalEndpoint,
+				Endpoint:         member.Endpoint,
+				PKIid:            id.ID,
+			}
+			req := d.createMembershipRequest(id.SelfOrg).NoopSign()
+			req.Nonce = util.RandomUInt64()
+			req.NoopSign()
+			go d.sendUntilAcked(peer, req)
 			return
 		}
+
 	}()
 }
 
+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) connect2BootstrapPeers(endpoints []string) {
 	if len(d.self.InternalEndpoint) == 0 {
 		d.logger.Panic("Internal endpoint is empty:", d.self.InternalEndpoint)
@@ -319,7 +342,7 @@ func (d *gossipDiscoveryImpl) handleMsgFromComm(m *proto.SignedGossipMessage) {
 		// Sending a membership response to a peer may block this routine
 		// in case the sending is deliberately slow (i.e attack).
 		// will keep this async until I'll write a timeout detector in the comm layer
-		go d.sendMemResponse(selfInfoGossipMsg.GetAliveMsg().Membership, internalEndpoint)
+		go d.sendMemResponse(selfInfoGossipMsg.GetAliveMsg().Membership, internalEndpoint, m.Nonce)
 		return
 	}
 
@@ -335,6 +358,7 @@ func (d *gossipDiscoveryImpl) handleMsgFromComm(m *proto.SignedGossipMessage) {
 	}
 
 	if memResp := m.GetMemRes(); memResp != nil {
+		d.pubsub.Publish(fmt.Sprintf("%d", m.Nonce), m.Nonce)
 		for _, env := range memResp.Alive {
 			am, err := env.ToGossipMessage()
 			if err != nil {
@@ -378,7 +402,7 @@ func (d *gossipDiscoveryImpl) handleMsgFromComm(m *proto.SignedGossipMessage) {
 	}
 }
 
-func (d *gossipDiscoveryImpl) sendMemResponse(targetMember *proto.Member, internalEndpoint string) {
+func (d *gossipDiscoveryImpl) sendMemResponse(targetMember *proto.Member, internalEndpoint string, nonce uint64) {
 	d.logger.Debug("Entering", targetMember)
 
 	targetPeer := &NetworkMember{
@@ -400,7 +424,7 @@ func (d *gossipDiscoveryImpl) sendMemResponse(targetMember *proto.Member, intern
 
 	d.comm.SendToPeer(targetPeer, (&proto.GossipMessage{
 		Tag:   proto.GossipMessage_EMPTY,
-		Nonce: uint64(0),
+		Nonce: nonce,
 		Content: &proto.GossipMessage_MemRes{
 			MemRes: memResp,
 		},

gossip/discovery/discovery_test.go

@@ -392,7 +392,9 @@ func TestConnect(t *testing.T) {
 		j := (i + 1) % 10
 		endpoint := fmt.Sprintf("localhost:%d", 7611+j)
 		netMember2Connect2 := NetworkMember{Endpoint: endpoint, PKIid: []byte(endpoint)}
-		inst.Connect(netMember2Connect2, func() bool { return false })
+		inst.Connect(netMember2Connect2, func() (identification *PeerIdentification, err error) {
+			return &PeerIdentification{SelfOrg: false, ID: nil}, nil
+		})
 	}
 
 	time.Sleep(time.Second * 3)