fix: solve potential deadlock for bsc p2p protocol handshake

eth/handler_bsc.go

@@ -17,21 +17,6 @@ func (h *bscHandler) Chain() *core.BlockChain { return h.chain }
 
 // RunPeer is invoked when a peer joins on the `bsc` protocol.
 func (h *bscHandler) RunPeer(peer *bsc.Peer, hand bsc.Handler) error {
-	if err := peer.Handshake(); err != nil {
-		// ensure that waitBscExtension receives the exit signal normally
-		// otherwise, can't graceful shutdown
-		ps := h.peers
-		id := peer.ID()
-
-		// Ensure nobody can double connect
-		ps.lock.Lock()
-		if wait, ok := ps.bscWait[id]; ok {
-			delete(ps.bscWait, id)
-			wait <- peer
-		}
-		ps.lock.Unlock()
-		return err
-	}
 	return (*handler)(h).runBscExtension(peer, hand)
 }
 

eth/handler_bsc_test.go

@@ -110,32 +110,32 @@ func testSendVotes(t *testing.T, protocol uint) {
 	defer localBsc.Close()
 	defer remoteBsc.Close()
 
-	go func(p *bsc.Peer) {
-		(*bscHandler)(handler.handler).RunPeer(p, func(peer *bsc.Peer) error {
-			return bsc.Handle((*bscHandler)(handler.handler), peer)
-		})
-	}(localBsc)
-
-	time.Sleep(200 * time.Millisecond)
-	remoteBsc.Handshake()
-
-	time.Sleep(200 * time.Millisecond)
 	go func(p *eth.Peer) {
 		handler.handler.runEthPeer(p, func(peer *eth.Peer) error {
 			return eth.Handle((*ethHandler)(handler.handler), peer)
 		})
 	}(localEth)
 
+	time.Sleep(200 * time.Millisecond)
+	go func(p *bsc.Peer) {
+		(*bscHandler)(handler.handler).RunPeer(p, func(peer *bsc.Peer) error {
+			return bsc.Handle((*bscHandler)(handler.handler), peer)
+		})
+	}(localBsc)
+
 	// Run the handshake locally to avoid spinning up a source handler
 	var (
 		genesis = handler.chain.Genesis()
 		head    = handler.chain.CurrentBlock()
 		td      = handler.chain.GetTd(head.Hash(), head.NumberU64())
 	)
-	time.Sleep(200 * time.Millisecond)
 	if err := remoteEth.Handshake(1, td, head.Hash(), genesis.Hash(), forkid.NewIDWithChain(handler.chain), forkid.NewFilter(handler.chain), nil); err != nil {
-		t.Fatalf("failed to run protocol handshake: %d", err)
+		t.Fatalf("failed to run eth protocol handshake: %d", err)
 	}
+	if err := remoteBsc.Handshake(); err != nil {
+		t.Fatalf("failed to run bsc protocol handshake: %d", err)
+	}
+
 	// After the handshake completes, the source handler should stream the sink
 	// the votes, subscribe to all inbound network events
 	backend := new(testBscHandler)
@@ -221,31 +221,30 @@ func testRecvVotes(t *testing.T, protocol uint) {
 	defer localBsc.Close()
 	defer remoteBsc.Close()
 
-	go func(p *bsc.Peer) {
-		(*bscHandler)(handler.handler).RunPeer(p, func(peer *bsc.Peer) error {
-			return bsc.Handle((*bscHandler)(handler.handler), peer)
-		})
-	}(localBsc)
-
-	time.Sleep(200 * time.Millisecond)
-	remoteBsc.Handshake()
-
-	time.Sleep(200 * time.Millisecond)
 	go func(p *eth.Peer) {
 		handler.handler.runEthPeer(p, func(peer *eth.Peer) error {
 			return eth.Handle((*ethHandler)(handler.handler), peer)
 		})
 	}(localEth)
 
+	time.Sleep(200 * time.Millisecond)
+	go func(p *bsc.Peer) {
+		(*bscHandler)(handler.handler).RunPeer(p, func(peer *bsc.Peer) error {
+			return bsc.Handle((*bscHandler)(handler.handler), peer)
+		})
+	}(localBsc)
+
 	// Run the handshake locally to avoid spinning up a source handler
 	var (
 		genesis = handler.chain.Genesis()
 		head    = handler.chain.CurrentBlock()
 		td      = handler.chain.GetTd(head.Hash(), head.NumberU64())
 	)
-	time.Sleep(200 * time.Millisecond)
 	if err := remoteEth.Handshake(1, td, head.Hash(), genesis.Hash(), forkid.NewIDWithChain(handler.chain), forkid.NewFilter(handler.chain), nil); err != nil {
-		t.Fatalf("failed to run protocol handshake: %d", err)
+		t.Fatalf("failed to run eth protocol handshake: %d", err)
+	}
+	if err := remoteBsc.Handshake(); err != nil {
+		t.Fatalf("failed to run bsc protocol handshake: %d", err)
 	}
 
 	votesCh := make(chan core.NewVoteEvent)

eth/peerset.go

@@ -458,7 +458,12 @@ func (ps *peerSet) registerPeer(peer *eth.Peer, ext *snap.Peer, diffExt *diff.Pe
 		eth.trustExt = &trustPeer{trustExt}
 	}
 	if bscExt != nil {
-		eth.bscExt = &bscPeer{bscExt}
+		if err := bscExt.Handshake(); err == nil {
+			eth.bscExt = &bscPeer{bscExt}
+		} else {
+			peer.Log().Debug("bsc protocol handshake failed", "err", err)
+			return err
+		}
 	}
 	ps.peers[id] = eth
 	return nil

eth/protocols/bsc/handler.go

@@ -1,6 +1,7 @@
 package bsc
 
 import (
+	"errors"
 	"fmt"
 	"time"
 
@@ -11,6 +12,17 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/enr"
 )
 
+const (
+	// handshakeWaitTimeout is the maximum allowed time for the extension waiting to
+	// complete handshake before dropping the connection as malicious.
+	handshakeWaitTimeout = 20 * time.Second
+)
+
+var (
+	// errHandshakeWaitTimeout is returned if a peer waits handshake for too long
+	errHandshakeWaitTimeout = errors.New("peer handshake wait timeout")
+)
+
 // Handler is a callback to invoke from an outside runner after the boilerplate
 // exchanges have passed.
 type Handler func(peer *Peer) error
@@ -74,6 +86,11 @@ func MakeProtocols(backend Backend, dnsdisc enode.Iterator) []p2p.Protocol {
 // Handle is the callback invoked to manage the life cycle of a `bsc` peer.
 // When this function terminates, the peer is disconnected.
 func Handle(backend Backend, peer *Peer) error {
+	select {
+	case <-peer.handshaked:
+	case <-time.After(handshakeWaitTimeout):
+		return errHandshakeWaitTimeout
+	}
 	for {
 		if err := handleMessage(backend, peer); err != nil {
 			peer.Log().Debug("Message handling failed in `bsc`", "err", err)

eth/protocols/bsc/handshake.go

@@ -42,6 +42,7 @@ func (p *Peer) Handshake() error {
 			return p2p.DiscReadTimeout
 		}
 	}
+	p.handshaked <- struct{}{}
 	return nil
 }
 

eth/protocols/bsc/peer.go

@@ -34,6 +34,8 @@ type Peer struct {
 	version   uint              // Protocol version negotiated
 	logger    log.Logger        // Contextual logger with the peer id injected
 	term      chan struct{}     // Termination channel to stop the broadcasters
+
+	handshaked chan struct{} // channel to start handleMessage after handshake
 }
 
 // NewPeer create a wrapper for a network connection and negotiated protocol
@@ -49,6 +51,7 @@ func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
 		version:       version,
 		logger:        log.New("peer", id[:8]),
 		term:          make(chan struct{}),
+		handshaked:    make(chan struct{}, 1), //asynchronous for handshake only once
 	}
 	go peer.broadcastVotes()
 	return peer