improve ethereum crawling

Author: dennis-tra

pkg/discv5/crawler.go

@@ -2,11 +2,8 @@ package discv5
 
 import (
 	"context"
-	"encoding/binary"
-	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"math/bits"
 	"strings"
 	"sync"
 	"time"
@@ -93,24 +90,26 @@ func (c *Crawler) PeerProperties(node *enode.Node) json.RawMessage {
 
 	var enrEntryEth2 ENREntryEth2
 	if err := node.Load(&enrEntryEth2); err == nil {
-		if beaconData, err := enrEntryEth2.Data(); err == nil {
-			properties["fork_digest"] = beaconData.ForkDigest.String()
-			properties["next_fork_version"] = beaconData.NextForkVersion.String()
-			properties["next_fork_epoch"] = beaconData.NextForkEpoch.String()
-		}
+		properties["fork_digest"] = enrEntryEth2.ForkDigest.String()
+		properties["next_fork_version"] = enrEntryEth2.NextForkVersion.String()
+		properties["next_fork_epoch"] = enrEntryEth2.NextForkEpoch.String()
 	}
 
 	var enrEntryAttnets ENREntryAttnets
 	if err := node.Load(&enrEntryAttnets); err == nil {
-		rawInt := binary.BigEndian.Uint64(enrEntryAttnets)
-		properties["attnets_num"] = bits.OnesCount64(rawInt)
-		properties["attnets"] = hex.EncodeToString(enrEntryAttnets)
+		properties["attnets_num"] = enrEntryAttnets.AttnetsNum
+		properties["attnets"] = enrEntryAttnets.Attnest
 	}
 
 	var enrEntrySyncCommsSubnet ENREntrySyncCommsSubnet
 	if err := node.Load(&enrEntrySyncCommsSubnet); err == nil {
-		// check out https://github.com/prysmaticlabs/prysm/blob/203dc5f63b060821c2706f03a17d66b3813c860c/beacon-chain/p2p/subnets.go#L221
-		properties["syncnets"] = hex.EncodeToString(enrEntrySyncCommsSubnet)
+		properties["syncnets"] = enrEntrySyncCommsSubnet.SyncNets
+	}
+
+	var enrEntryOpStack ENREntryOpStack
+	if err := node.Load(&enrEntryOpStack); err == nil {
+		properties["opstack_chain_id"] = enrEntryOpStack.ChainID
+		properties["opstack_version"] = enrEntryOpStack.Version
 	}
 
 	data, err := json.Marshal(properties)
@@ -202,17 +201,31 @@ func (c *Crawler) connect(ctx context.Context, pi peer.AddrInfo) error {
 	if len(pi.Addrs) == 0 {
 		metrics.VisitErrorsCount.With(metrics.CrawlLabel).Inc()
 		return fmt.Errorf("skipping node as it has no public IP address") // change knownErrs map if changing this msg
+	} else if len(pi.Addrs) == 1 {
+	}
+
+	dialAddrInfo := peer.AddrInfo{
+		ID:    pi.ID,
+		Addrs: ensureTCPAddr(pi.Addrs),
+	}
+
+	replaced := false
+	if len(dialAddrInfo.Addrs) != len(pi.Addrs) {
+		replaced = true
 	}
 
 	retry := 0
 	maxRetries := 1
 	for {
 		timeout := time.Duration(c.cfg.DialTimeout.Nanoseconds() / int64(retry+1))
 		timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
-		err := c.host.Connect(timeoutCtx, pi)
+		err := c.host.Connect(timeoutCtx, dialAddrInfo)
 		cancel()
 
 		if err == nil {
+			if replaced {
+				log.WithField("remoteID", pi.ID.ShortString()).Errorln("hat was gebracht!")
+			}
 			return nil
 		}
 
@@ -241,6 +254,52 @@ func (c *Crawler) connect(ctx context.Context, pi peer.AddrInfo) error {
 	}
 }
 
+func ensureTCPAddr(maddrs []ma.Multiaddr) []ma.Multiaddr {
+	for _, maddr := range maddrs {
+		if _, err := maddr.ValueForProtocol(ma.P_TCP); err == nil {
+			return maddrs
+		}
+	}
+
+	newMaddrs := make([]ma.Multiaddr, 0, len(maddrs)+1)
+
+	for i, maddr := range maddrs {
+		newMaddrs = append(newMaddrs, maddr)
+
+		udp, err := maddr.ValueForProtocol(ma.P_UDP)
+		if err != nil {
+			continue
+		}
+
+		ip := ""
+		ip4, err := maddr.ValueForProtocol(ma.P_IP4)
+		if err != nil {
+			ip6, err := maddr.ValueForProtocol(ma.P_IP6)
+			if err != nil {
+				continue
+			}
+			ip = "/ip6/" + ip6
+		} else {
+			ip = "/ip4/" + ip4
+		}
+
+		tcpMaddr, err := ma.NewMultiaddr(ip + "/tcp/" + udp)
+		if err != nil {
+			continue
+		}
+
+		for _, remaining := range maddrs[i+1:] {
+			newMaddrs = append(newMaddrs, remaining)
+		}
+
+		newMaddrs = append(newMaddrs, tcpMaddr)
+
+		return newMaddrs
+	}
+
+	return maddrs
+}
+
 // identifyWait waits until any connection to a peer passed the Identify
 // exchange successfully or all identification attempts have failed.
 // The call to IdentifyWait returns immediately if the connection was

pkg/discv5/crawler_test.go

@@ -0,0 +1,60 @@
+package discv5
+
+import (
+	"testing"
+
+	ma "github.com/multiformats/go-multiaddr"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/dennis-tra/nebula-crawler/pkg/nebtest"
+)
+
+func Test_ensureTCPAddr(t *testing.T) {
+	tests := []struct {
+		name   string
+		maddrs []ma.Multiaddr
+		want   []ma.Multiaddr
+	}{
+		{
+			name:   "empty",
+			maddrs: []ma.Multiaddr{},
+			want:   []ma.Multiaddr{},
+		},
+		{
+			name: "tcp exists",
+			maddrs: []ma.Multiaddr{
+				nebtest.MustMultiaddr(t, "/ip4/123.1.1.1/udp/1234"),
+				nebtest.MustMultiaddr(t, "/ip4/123.1.1.1/tcp/1234"),
+			},
+			want: []ma.Multiaddr{
+				nebtest.MustMultiaddr(t, "/ip4/123.1.1.1/udp/1234"),
+				nebtest.MustMultiaddr(t, "/ip4/123.1.1.1/tcp/1234"),
+			},
+		},
+		{
+			name: "single udp ip4",
+			maddrs: []ma.Multiaddr{
+				nebtest.MustMultiaddr(t, "/ip4/123.1.1.1/udp/1234"),
+			},
+			want: []ma.Multiaddr{
+				nebtest.MustMultiaddr(t, "/ip4/123.1.1.1/udp/1234"),
+				nebtest.MustMultiaddr(t, "/ip4/123.1.1.1/tcp/1234"),
+			},
+		},
+		{
+			name: "single udp ip6",
+			maddrs: []ma.Multiaddr{
+				nebtest.MustMultiaddr(t, "/ip6/::1/udp/1234"),
+			},
+			want: []ma.Multiaddr{
+				nebtest.MustMultiaddr(t, "/ip6/::1/udp/1234"),
+				nebtest.MustMultiaddr(t, "/ip6/::1/tcp/1234"),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equalf(t, tt.want, ensureTCPAddr(tt.maddrs), "ensureTCPAddr(%v)", tt.maddrs)
+		})
+	}
+}

pkg/discv5/dialer.go

@@ -20,7 +20,8 @@ type Dialer struct {
 
 var _ core.Worker[PeerInfo, core.DialResult[PeerInfo]] = (*Dialer)(nil)
 
-// Work TODO
+// Work takes the PeerInfo object and tries to figure out if the peer is
+// still online.
 func (d *Dialer) Work(ctx context.Context, task PeerInfo) (core.DialResult[PeerInfo], error) {
 	// Creating log entry
 	logEntry := log.WithFields(log.Fields{

pkg/discv5/driver_crawler.go

@@ -5,6 +5,7 @@ import (
 	"crypto/elliptic"
 	crand "crypto/rand"
 	"fmt"
+	"io"
 	"net"
 	"time"
 
@@ -15,6 +16,7 @@ import (
 	"github.com/libp2p/go-libp2p"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
 	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
@@ -157,6 +159,7 @@ func NewCrawlDriver(dbc db.Client, crawl *models.Crawl, cfg *CrawlDriverConfig)
 
 	privBytes := elliptic.Marshal(ethcrypto.S256(), ecdsaKey.X, ecdsaKey.Y)
 	secpKey := (*crypto.Secp256k1PrivateKey)(secp256k1v4.PrivKeyFromBytes(privBytes))
+
 	// Initialize a single libp2p node that's shared between all crawlers.
 	// Context: https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/p2p-interface.md#network-fundamentals
 	h, err := libp2p.New(
@@ -173,6 +176,19 @@ func NewCrawlDriver(dbc db.Client, crawl *models.Crawl, cfg *CrawlDriverConfig)
 		return nil, fmt.Errorf("new libp2p host: %w", err)
 	}
 
+	// According to Diva, these are required protocols. Some of them are just
+	// assumed to be required. We just read from the stream indefinitely to
+	// gain time for the identify exchange to finish. We just pretend to support
+	// these protocols and keep the stream busy until we have gathered all the
+	// information we were interested in. This includes the agend version and
+	// all supported protocols.
+	h.SetStreamHandler("/eth2/beacon_chain/req/ping/1/ssz_snappy", func(s network.Stream) { io.ReadAll(s) })
+	h.SetStreamHandler("/eth2/beacon_chain/req/status/1/ssz_snappy", func(s network.Stream) { io.ReadAll(s) })
+	h.SetStreamHandler("/eth2/beacon_chain/req/metadata/1/ssz_snappy", func(s network.Stream) { io.ReadAll(s) })
+	h.SetStreamHandler("/eth2/beacon_chain/req/metadata/2/ssz_snappy", func(s network.Stream) { io.ReadAll(s) })
+	h.SetStreamHandler("/eth2/beacon_chain/req/goodbye/1/ssz_snappy", func(s network.Stream) { io.ReadAll(s) })
+	h.SetStreamHandler("/meshsub/1.1.0", func(s network.Stream) { io.ReadAll(s) }) // https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/p2p-interface.md#the-gossip-domain-gossipsub
+
 	log.WithField("peerID", h.ID().String()).Infoln("Started libp2p host")
 
 	nodesMap := map[enode.ID]*enode.Node{}

pkg/discv5/driver_crawler_test.go

@@ -0,0 +1,53 @@
+package discv5
+
+import (
+	"testing"
+
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewPeerInfo(t *testing.T) {
+	type args struct {
+		node *enode.Node
+	}
+	tests := []struct {
+		name   string
+		enr    string
+		peerID string
+		maddrs []string
+	}{
+		{
+			name:   "bootstrap 1",
+			enr:    "enr:-Ku4QEWzdnVtXc2Q0ZVigfCGggOVB2Vc1ZCPEc6j21NIFLODSJbvNaef1g4PxhPwl_3kax86YPheFUSLXPRs98vvYsoBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpC1MD8qAAAAAP__________gmlkgnY0gmlwhDZBrP2Jc2VjcDI1NmsxoQM6jr8Rb1ktLEsVcKAPa08wCsKUmvoQ8khiOl_SLozf9IN1ZHCCIyg",
+			peerID: "16Uiu2HAmGbaF2hpC4iumV7eWiK51nWmsZf7mK2R9nCM6rrvRQCwu",
+			maddrs: []string{
+				"/ip4/54.65.172.253/udp/9000",
+			},
+		},
+		{
+			name:   "bootstrap 2",
+			enr:    "enr:-LK4QKWrXTpV9T78hNG6s8AM6IO4XH9kFT91uZtFg1GcsJ6dKovDOr1jtAAFPnS2lvNltkOGA9k29BUN7lFh_sjuc9QBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpC1MD8qAAAAAP__________gmlkgnY0gmlwhANAdd-Jc2VjcDI1NmsxoQLQa6ai7y9PMN5hpLe5HmiJSlYzMuzP7ZhwRiwHvqNXdoN0Y3CCI4yDdWRwgiOM",
+			peerID: "16Uiu2HAm9TFzGRAgJKVyfZ54nDbCYNwEAFnpz6ybjM7KaVPLJZpy",
+			maddrs: []string{
+				"/ip4/3.64.117.223/udp/9100",
+				"/ip4/3.64.117.223/tcp/9100",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			node, err := enode.Parse(enode.ValidSchemes, tt.enr)
+			require.NoError(t, err)
+
+			got, err := NewPeerInfo(node)
+			assert.Nil(t, err)
+			assert.Equal(t, tt.peerID, got.peerID.String())
+
+			for i, maddr := range got.maddrs {
+				assert.Equal(t, tt.maddrs[i], maddr.String())
+			}
+		})
+	}
+}