Add the Ability for Prysm To Handle Trusted Peers (#12492)

* add all changes

* add to peers to watch

* add tests

* Update beacon-chain/p2p/peers/peerdata/store_test.go

* radek's review

---------

Co-authored-by: Radosław Kapka <rkapka@wp.pl>
Co-authored-by: james-prysm <90280386+james-prysm@users.noreply.github.com>

beacon-chain/p2p/discovery.go

@@ -460,6 +460,19 @@ func convertToUdpMultiAddr(node *enode.Node) ([]ma.Multiaddr, error) {
 	return addresses, nil
 }
 
+func peerIdsFromMultiAddrs(addrs []ma.Multiaddr) []peer.ID {
+	peers := []peer.ID{}
+	for _, a := range addrs {
+		info, err := peer.AddrInfoFromP2pAddr(a)
+		if err != nil {
+			log.WithError(err).Errorf("Could not derive peer info from multiaddress %s", a.String())
+			continue
+		}
+		peers = append(peers, info.ID)
+	}
+	return peers
+}
+
 func multiAddrFromString(address string) (ma.Multiaddr, error) {
 	return ma.NewMultiaddr(address)
 }

beacon-chain/p2p/peers/peerdata/store.go

@@ -38,9 +38,10 @@ type StoreConfig struct {
 // the mutex when accessing data.
 type Store struct {
 	sync.RWMutex
-	ctx    context.Context
-	config *StoreConfig
-	peers  map[peer.ID]*PeerData
+	ctx          context.Context
+	config       *StoreConfig
+	peers        map[peer.ID]*PeerData
+	trustedPeers map[peer.ID]bool
 }
 
 // PeerData aggregates protocol and application level info about a single peer.
@@ -69,9 +70,10 @@ type PeerData struct {
 // NewStore creates new peer data store.
 func NewStore(ctx context.Context, config *StoreConfig) *Store {
 	return &Store{
-		ctx:    ctx,
-		config: config,
-		peers:  make(map[peer.ID]*PeerData),
+		ctx:          ctx,
+		config:       config,
+		peers:        make(map[peer.ID]*PeerData),
+		trustedPeers: make(map[peer.ID]bool),
 	}
 }
 
@@ -105,12 +107,25 @@ func (s *Store) DeletePeerData(pid peer.ID) {
 	delete(s.peers, pid)
 }
 
+// SetTrustedPeers sets our desired trusted peer set.
+func (s *Store) SetTrustedPeers(peers []peer.ID) {
+	for _, p := range peers {
+		s.trustedPeers[p] = true
+	}
+}
+
 // Peers returns map of peer data objects.
 // Important: it is assumed that store mutex is locked when calling this method.
 func (s *Store) Peers() map[peer.ID]*PeerData {
 	return s.peers
 }
 
+// IsTrustedPeer checks that the provided peer
+// is in our trusted peer set.
+func (s *Store) IsTrustedPeer(p peer.ID) bool {
+	return s.trustedPeers[p]
+}
+
 // Config exposes store configuration params.
 func (s *Store) Config() *StoreConfig {
 	return s.config

beacon-chain/p2p/peers/peerdata/store_test.go

@@ -80,3 +80,20 @@ func TestStore_PeerDataGetOrCreate(t *testing.T) {
 	assert.Equal(t, uint64(0), peerData.ProcessedBlocks)
 	require.Equal(t, 1, len(store.Peers()))
 }
+
+func TestStore_TrustedPeers(t *testing.T) {
+	store := peerdata.NewStore(context.Background(), &peerdata.StoreConfig{
+		MaxPeers: 12,
+	})
+
+	pid1 := peer.ID("00001")
+	pid2 := peer.ID("00002")
+	pid3 := peer.ID("00003")
+
+	tPeers := []peer.ID{pid1, pid2, pid3}
+	store.SetTrustedPeers(tPeers)
+
+	assert.Equal(t, true, store.IsTrustedPeer(pid1))
+	assert.Equal(t, true, store.IsTrustedPeer(pid2))
+	assert.Equal(t, true, store.IsTrustedPeer(pid3))
+}

beacon-chain/p2p/peers/status.go

@@ -335,6 +335,10 @@ func (p *Status) IsBad(pid peer.ID) bool {
 
 // isBad is the lock-free version of IsBad.
 func (p *Status) isBad(pid peer.ID) bool {
+	// Do not disconnect from trusted peers.
+	if p.store.IsTrustedPeer(pid) {
+		return false
+	}
 	return p.isfromBadIP(pid) || p.scorers.IsBadPeerNoLock(pid)
 }
 
@@ -769,7 +773,7 @@ func (p *Status) PeersToPrune() []peer.ID {
 	// Select connected and inbound peers to prune.
 	for pid, peerData := range p.store.Peers() {
 		if peerData.ConnState == PeerConnected &&
-			peerData.Direction == network.DirInbound {
+			peerData.Direction == network.DirInbound && !p.store.IsTrustedPeer(pid) {
 			peersToPrune = append(peersToPrune, &peerResp{
 				pid:   pid,
 				score: p.scorers.ScoreNoLock(pid),
@@ -835,7 +839,7 @@ func (p *Status) deprecatedPeersToPrune() []peer.ID {
 	// Select connected and inbound peers to prune.
 	for pid, peerData := range p.store.Peers() {
 		if peerData.ConnState == PeerConnected &&
-			peerData.Direction == network.DirInbound {
+			peerData.Direction == network.DirInbound && !p.store.IsTrustedPeer(pid) {
 			peersToPrune = append(peersToPrune, &peerResp{
 				pid:     pid,
 				badResp: peerData.BadResponses,
@@ -900,6 +904,14 @@ func (p *Status) ConnectedPeerLimit() uint64 {
 	return uint64(maxLim) - maxLimitBuffer
 }
 
+// SetTrustedPeers sets our trusted peer set into
+// our peerstore.
+func (p *Status) SetTrustedPeers(peers []peer.ID) {
+	p.store.Lock()
+	defer p.store.Unlock()
+	p.store.SetTrustedPeers(peers)
+}
+
 // this method assumes the store lock is acquired before
 // executing the method.
 func (p *Status) isfromBadIP(pid peer.ID) bool {

beacon-chain/p2p/peers/status_test.go

@@ -755,6 +755,78 @@ func TestPrunePeers(t *testing.T) {
 	}
 }
 
+func TestPrunePeers_TrustedPeers(t *testing.T) {
+	p := peers.NewStatus(context.Background(), &peers.StatusConfig{
+		PeerLimit: 30,
+		ScorerParams: &scorers.Config{
+			BadResponsesScorerConfig: &scorers.BadResponsesScorerConfig{
+				Threshold: 1,
+			},
+		},
+	})
+
+	for i := 0; i < 15; i++ {
+		// Peer added to peer handler.
+		createPeer(t, p, nil, network.DirOutbound, peerdata.PeerConnectionState(ethpb.ConnectionState_CONNECTED))
+	}
+	// Assert there are no prunable peers.
+	peersToPrune := p.PeersToPrune()
+	assert.Equal(t, 0, len(peersToPrune))
+
+	for i := 0; i < 18; i++ {
+		// Peer added to peer handler.
+		createPeer(t, p, nil, network.DirInbound, peerdata.PeerConnectionState(ethpb.ConnectionState_CONNECTED))
+	}
+
+	// Assert there are the correct prunable peers.
+	peersToPrune = p.PeersToPrune()
+	assert.Equal(t, 3, len(peersToPrune))
+
+	// Add in more peers.
+	for i := 0; i < 13; i++ {
+		// Peer added to peer handler.
+		createPeer(t, p, nil, network.DirInbound, peerdata.PeerConnectionState(ethpb.ConnectionState_CONNECTED))
+	}
+
+	trustedPeers := []peer.ID{}
+	// Set up bad scores for inbound peers.
+	inboundPeers := p.InboundConnected()
+	for i, pid := range inboundPeers {
+		modulo := i % 5
+		// Increment bad scores for peers.
+		for j := 0; j < modulo; j++ {
+			p.Scorers().BadResponsesScorer().Increment(pid)
+		}
+		if modulo == 4 {
+			trustedPeers = append(trustedPeers, pid)
+		}
+	}
+	p.SetTrustedPeers(trustedPeers)
+	// Assert all peers more than max are prunable.
+	peersToPrune = p.PeersToPrune()
+	assert.Equal(t, 16, len(peersToPrune))
+
+	// Check that trusted peers are not pruned.
+	for _, pid := range peersToPrune {
+		for _, tPid := range trustedPeers {
+			assert.NotEqual(t, pid.String(), tPid.String())
+		}
+	}
+	for _, pid := range peersToPrune {
+		dir, err := p.Direction(pid)
+		require.NoError(t, err)
+		assert.Equal(t, network.DirInbound, dir)
+	}
+
+	// Ensure it is in the descending order.
+	currScore := p.Scorers().Score(peersToPrune[0])
+	for _, pid := range peersToPrune {
+		score := p.Scorers().BadResponsesScorer().Score(pid)
+		assert.Equal(t, true, currScore >= score)
+		currScore = score
+	}
+}
+
 func TestStatus_BestPeer(t *testing.T) {
 	type peerConfig struct {
 		headSlot       primitives.Slot

beacon-chain/p2p/service.go

@@ -210,6 +210,10 @@ func (s *Service) Start() {
 		if err != nil {
 			log.WithError(err).Error("Could not connect to static peer")
 		}
+		// Set trusted peers for those that are provided as static addresses.
+		pids := peerIdsFromMultiAddrs(addrs)
+		s.peers.SetTrustedPeers(pids)
+		peersToWatch = append(peersToWatch, s.cfg.StaticPeers...)
 		s.connectWithAllPeers(addrs)
 	}
 	// Initialize metadata according to the

cmd/flags.go

@@ -93,7 +93,7 @@ var (
 	// StaticPeers specifies a set of peers to connect to explicitly.
 	StaticPeers = &cli.StringSliceFlag{
 		Name:  "peer",
-		Usage: "Connect with this peer. This flag may be used multiple times.",
+		Usage: "Connect with this peer, this flag may be used multiple times. This peer is recognized as a trusted peer.",
 	}
 	// BootstrapNode tells the beacon node which bootstrap node to connect to
 	BootstrapNode = &cli.StringSliceFlag{