feat(header|p2p): add blacklisting to peerTracker

header/p2p/exchange.go

@@ -11,6 +11,7 @@ import (
  "github.com/libp2p/go-libp2p-core/host"
  "github.com/libp2p/go-libp2p-core/peer"
  "github.com/libp2p/go-libp2p-core/protocol"
+ "github.com/libp2p/go-libp2p/p2p/net/conngater"
  tmbytes "github.com/tendermint/tendermint/libs/bytes"
 
  "github.com/celestiaorg/celestia-node/header"
@@ -42,6 +43,7 @@ func NewExchange(
  host host.Host,
  peers peer.IDSlice,
  protocolSuffix string,
+ connGater *conngater.BasicConnectionGater,
  opts ...Option[ClientParameters],
 ) (*Exchange, error) {
  params := DefaultClientParameters()
@@ -60,6 +62,7 @@ func NewExchange(
  trustedPeers: peers,
  peerTracker: newPeerTracker(
  host,
+ connGater,
  params.MaxAwaitingTime,
  params.DefaultScore,
  params.MaxPeerTrackerSize,

header/p2p/exchange_test.go

@@ -6,8 +6,11 @@ import (
  "testing"
  "time"
 
+ "github.com/ipfs/go-datastore"
+ "github.com/ipfs/go-datastore/sync"
  libhost "github.com/libp2p/go-libp2p-core/host"
  "github.com/libp2p/go-libp2p-core/peer"
+ "github.com/libp2p/go-libp2p/p2p/net/conngater"
  mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
@@ -82,8 +85,10 @@ func TestExchange_RequestFullRangeHeaders(t *testing.T) {
  totalAmount := 80
  store := createStore(t, totalAmount)
  protocolSuffix := "private"
+ connGater, err := conngater.NewBasicConnectionGater(sync.MutexWrap(datastore.NewMapDatastore()))
+ require.NoError(t, err)
  // create new exchange
- exchange, err := NewExchange(hosts[len(hosts)-1], []peer.ID{}, protocolSuffix)
+ exchange, err := NewExchange(hosts[len(hosts)-1], []peer.ID{}, protocolSuffix, connGater)
  require.NoError(t, err)
  exchange.Params.MaxHeadersPerRequest = 10
  exchange.ctx, exchange.cancel = context.WithCancel(context.Background())
@@ -288,8 +293,9 @@ func createP2PExAndServer(t *testing.T, host, tpeer libhost.Host) (header.Exchan
  require.NoError(t, err)
  err = serverSideEx.Start(context.Background())
  require.NoError(t, err)
-
- ex, err := NewExchange(host, []peer.ID{tpeer.ID()}, "private")
+ connGater, err := conngater.NewBasicConnectionGater(sync.MutexWrap(datastore.NewMapDatastore()))
+ require.NoError(t, err)
+ ex, err := NewExchange(host, []peer.ID{tpeer.ID()}, "private", connGater)
  require.NoError(t, err)
  ex.peerTracker.trackedPeers[tpeer.ID()] = &peerStat{peerID: tpeer.ID()}
  require.NoError(t, ex.Start(context.Background()))

header/p2p/peer_tracker.go

@@ -9,13 +9,15 @@ import (
  "github.com/libp2p/go-libp2p-core/host"
  "github.com/libp2p/go-libp2p-core/network"
  "github.com/libp2p/go-libp2p-core/peer"
+ "github.com/libp2p/go-libp2p/p2p/net/conngater"
 )
 
 // gcCycle defines the duration after which the peerTracker starts removing peers.
 var gcCycle = time.Minute * 30
 
 type peerTracker struct {
- host host.Host
+ host host.Host
+ connGater *conngater.BasicConnectionGater
 
  peerLk sync.RWMutex
  // trackedPeers contains active peers that we can request to.
@@ -43,13 +45,15 @@ type peerTracker struct {
 
 func newPeerTracker(
  h host.Host,
+ connGater *conngater.BasicConnectionGater,
  maxAwaitingTime time.Duration,
  defaultScore float32,
  maxPeerTrackerSize int,
 ) *peerTracker {
  ctx, cancel := context.WithCancel(context.Background())
  return &peerTracker{
  host: h,
+ connGater: connGater,
  disconnectedPeers: make(map[peer.ID]*peerStat),
  trackedPeers: make(map[peer.ID]*peerStat),
  maxAwaitingTime: maxAwaitingTime,
@@ -190,3 +194,25 @@ func (p *peerTracker) stop() {
  <-p.done
  }
 }
+
+// blockPeer blocks a peer on the networking level and removes it from the local cache.
+func (p *peerTracker) blockPeer(pID peer.ID) {
+ // add peer to the blacklist, so we can't connect to it in the future.
+ err := p.connGater.BlockPeer(pID)
+ if err != nil {
+ log.Errorw("header/p2p: blocking peer failed", "err", err, "pID", pID)
+ }
+ // close connections to peer.
+ err = p.host.Network().ClosePeer(pID)
+ if err != nil {
+ log.Errorw("header/p2p: closing connection with peer failed", "err", err, "pID", pID)
+ }
+
+ log.Warnw("header/p2p: blocked peer", "pID", pID)
+
+ p.peerLk.Lock()
+ defer p.peerLk.Unlock()
+ // remove peer from cache.
+ delete(p.trackedPeers, pID)
+ delete(p.disconnectedPeers, pID)
+}

header/p2p/peer_tracker_test.go

@@ -4,15 +4,20 @@ import (
  "testing"
  "time"
 
+ "github.com/ipfs/go-datastore"
+ "github.com/ipfs/go-datastore/sync"
  "github.com/libp2p/go-libp2p-core/peer"
+ "github.com/libp2p/go-libp2p/p2p/net/conngater"
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
 )
 
 func TestPeerTracker_GC(t *testing.T) {
  h := createMocknet(t, 1)
  gcCycle = time.Millisecond * 200
- p := newPeerTracker(h[0], time.Millisecond*1, 1, 5)
+ connGater, err := conngater.NewBasicConnectionGater(sync.MutexWrap(datastore.NewMapDatastore()))
+ require.NoError(t, err)
+ p := newPeerTracker(h[0], connGater, time.Millisecond*1, 1, 5)
  pid1 := peer.ID("peer1")
  pid2 := peer.ID("peer2")
  pid3 := peer.ID("peer3")
@@ -31,3 +36,13 @@ func TestPeerTracker_GC(t *testing.T) {
  require.Nil(t, p.trackedPeers[pid1])
  require.Nil(t, p.disconnectedPeers[pid3])
 }
+
+func TestPeerTracker_BlockPeer(t *testing.T) {
+ h := createMocknet(t, 2)
+ connGater, err := conngater.NewBasicConnectionGater(sync.MutexWrap(datastore.NewMapDatastore()))
+ require.NoError(t, err)
+ p := newPeerTracker(h[0], connGater, time.Millisecond*1, 1, 5)
+ p.blockPeer(h[1].ID())
+ require.Len(t, connGater.ListBlockedPeers(), 1)
+ require.True(t, connGater.ListBlockedPeers()[0] == h[1].ID())
+}

nodebuilder/header/constructors.go

@@ -7,6 +7,7 @@ import (
  "github.com/libp2p/go-libp2p-core/host"
  "github.com/libp2p/go-libp2p-core/peer"
  "github.com/libp2p/go-libp2p-core/peerstore"
+ "github.com/libp2p/go-libp2p/p2p/net/conngater"
  "go.uber.org/fx"
 
  "github.com/celestiaorg/celestia-node/header"
@@ -32,13 +33,15 @@ func newP2PExchange(cfg Config) func(
  modp2p.Bootstrappers,
  modp2p.Network,
  host.Host,
+ *conngater.BasicConnectionGater,
  []p2p.Option[p2p.ClientParameters],
 ) (header.Exchange, error) {
  return func(
  lc fx.Lifecycle,
  bpeers modp2p.Bootstrappers,
  network modp2p.Network,
  host host.Host,
+ conngater *conngater.BasicConnectionGater,
  opts []p2p.Option[p2p.ClientParameters],
  ) (header.Exchange, error) {
  peers, err := cfg.trustedPeers(bpeers)
@@ -50,7 +53,7 @@ func newP2PExchange(cfg Config) func(
  ids[index] = peer.ID
  host.Peerstore().AddAddrs(peer.ID, peer.Addrs, peerstore.PermanentAddrTTL)
  }
- exchange, err := p2p.NewExchange(host, ids, string(network), opts...)
+ exchange, err := p2p.NewExchange(host, ids, string(network), conngater, opts...)
  if err != nil {
  return nil, err
  }

nodebuilder/header/module_test.go

@@ -5,6 +5,7 @@ import (
 
  "github.com/ipfs/go-datastore"
  "github.com/libp2p/go-libp2p"
+ "github.com/libp2p/go-libp2p/p2p/net/conngater"
  "github.com/stretchr/testify/require"
  "go.uber.org/fx"
  "go.uber.org/fx/fxtest"
@@ -60,6 +61,9 @@ func TestConstructModule_ExchangeParams(t *testing.T) {
  return datastore.NewMapDatastore()
  }),
  ConstructModule(node.Light, &cfg),
+ fx.Provide(func(b datastore.Batching) (*conngater.BasicConnectionGater, error) {
+ return conngater.NewBasicConnectionGater(b)
+ }),
  fx.Invoke(
  func(e header.Exchange, server *p2p.ExchangeServer) {
  ex := e.(*p2p.Exchange)