feat: initial seeds peering version (working)

for now, --seeds-peering is incompatible with --routing since we need to use the dht to find our peers from the same "seed ring". I will rework the routing option so that we can enable and disable routing v1 and dht independently.

keys.go

@@ -5,9 +5,11 @@ import (
 	crand "crypto/rand"
 	"crypto/sha256"
 	"errors"
+	"fmt"
 	"io"
 
 	libp2p "github.com/libp2p/go-libp2p/core/crypto"
+	peer "github.com/libp2p/go-libp2p/core/peer"
 	"github.com/mr-tron/base58"
 	"golang.org/x/crypto/hkdf"
 )
@@ -24,7 +26,7 @@ func newSeed() (string, error) {
 	return base58.Encode(bs), nil
 }
 
-// derive derives libp2p keys from a b58-encoded seed.
+// deriveKey derives libp2p keys from a b58-encoded seed.
 func deriveKey(b58secret string, info []byte) (libp2p.PrivKey, error) {
 	secret, err := base58.Decode(b58secret)
 	if err != nil {
@@ -43,3 +45,33 @@ func deriveKey(b58secret string, info []byte) (libp2p.PrivKey, error) {
 	key := ed25519.NewKeyFromSeed(keySeed)
 	return libp2p.UnmarshalEd25519PrivateKey(key)
 }
+
+// derivePeerIDs derives the peer IDs of all the peers with the same seed up to
+// maxIndex. Our peer ID (with index 'ourIndex') is not generated.
+func derivePeerIDs(seed string, ourIndex int, maxIndex int) ([]peer.ID, error) {
+	peerIDs := []peer.ID{}
+
+	for i := 0; i <= maxIndex; i++ {
+		if i == ourIndex {
+			continue
+		}
+
+		peerPriv, err := deriveKey(seed, deriveInfo(i))
+		if err != nil {
+			return nil, err
+		}
+
+		pid, err := peer.IDFromPrivateKey(peerPriv)
+		if err != nil {
+			return nil, err
+		}
+
+		peerIDs = append(peerIDs, pid)
+	}
+
+	return peerIDs, nil
+}
+
+func deriveInfo(index int) []byte {
+	return []byte(fmt.Sprintf("rainbow-%d", index))
+}

main.go

@@ -89,6 +89,18 @@ Generate an identity seed and launch a gateway:
 			EnvVars: []string{"RAINBOW_SEED_INDEX"},
 			Usage:   "Index to derivate the peerID (needs --seed)",
 		},
+		&cli.BoolFlag{
+			Name:    "seeds-peering",
+			Value:   false,
+			EnvVars: []string{"RAINBOW_SEEDS_PEERING"},
+			Usage:   "TODO (needs --seed and --seed-index). Automatically enables --dht-shared-host, incompatible with --routing",
+		},
+		&cli.UintFlag{
+			Name:    "seeds-peering-max-index",
+			Value:   100,
+			EnvVars: []string{"RAINBOW_SEEDS_PEERING_MAX_INDEX"},
+			Usage:   "TODO",
+		},
 		&cli.StringSliceFlag{
 			Name:    "gateway-domains",
 			Value:   cli.NewStringSlice(),
@@ -281,7 +293,7 @@ share the same seed as long as the indexes are different.
 		index := cctx.Int("seed-index")
 		if len(seed) > 0 && index >= 0 {
 			fmt.Println("Deriving identity from seed")
-			priv, err = deriveKey(seed, []byte(fmt.Sprintf("rainbow-%d", index)))
+			priv, err = deriveKey(seed, deriveInfo(index))
 		} else {
 			fmt.Println("Setting identity from libp2p.key")
 			keyFile := filepath.Join(secretsDir, "libp2p.key")
@@ -300,6 +312,37 @@ share the same seed as long as the indexes are different.
 			peeringAddrs = append(peeringAddrs, *ai)
 		}
 
+		if cctx.Bool("seeds-peering") {
+			if !cctx.IsSet("seed") || !cctx.IsSet("seed-index") {
+				return errors.New("--seed and --seed-index must be explicitly defined when --seeds-peering is enabled")
+			}
+
+			if cctx.String("dht-routing") == "off" {
+				return errors.New("--dht-routing=off is incompatible with --seeds-peering: DHT needs to run in order to be able to find peers")
+			}
+
+			// If seeds-peering is enabled, automatically use the same libp2p host
+			// for the DHT. This allows the peer information (id and addresses) to
+			// be discoverable via the DHT without having to create a new DHT client
+			// instance just for this purpose.
+			err = cctx.Set("dht-shared-host", "true")
+			if err != nil {
+				return err
+			}
+
+			maxIndex := cctx.Uint("seeds-peering-max-index")
+			peeringIDs, err := derivePeerIDs(seed, index, int(maxIndex))
+			if err != nil {
+				return err
+			}
+
+			for _, pid := range peeringIDs {
+				// The peering module will automatically perform lookups to find the
+				// addresses of the given peers.
+				peeringAddrs = append(peeringAddrs, peer.AddrInfo{ID: pid})
+			}
+		}
+
 		cfg := Config{
 			DataDir:                 ddir,
 			BlockstoreType:          cctx.String("blockstore"),

setup.go

@@ -47,6 +47,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/routing"
 	"github.com/libp2p/go-libp2p/p2p/net/connmgr"
+	"github.com/libp2p/go-libp2p/p2p/protocol/identify"
 	"go.opencensus.io/stats/view"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
@@ -67,6 +68,11 @@ const (
 	DHTOff         DHTRouting = "off"
 )
 
+func init() {
+	// Lets us discover our own public address with a single observation
+	identify.ActivationThresh = 1
+}
+
 type Node struct {
 	vs   routing.ValueStore
 	host host.Host
@@ -79,6 +85,7 @@ type Node struct {
 	resolver   resolver.Resolver
 
 	ns namesys.NameSystem
+	ps *peering.PeeringService
 
 	bwc *metrics.BandwidthCounter
 
@@ -291,8 +298,9 @@ func Setup(ctx context.Context, cfg Config, key crypto.PrivKey, dnsCache *cached
 		return nil, err
 	}
 
+	var ps *peering.PeeringService
 	if len(cfg.Peering) > 0 {
-		ps := peering.NewPeeringService(h)
+		ps = peering.NewPeeringService(h)
 		if err := ps.Start(); err != nil {
 			return nil, err
 		}

setup_test.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/crypto"
+	peer "github.com/libp2p/go-libp2p/core/peer"
+	"github.com/stretchr/testify/require"
+)
+
+func testSeedPeering(t *testing.T, n int) ([]crypto.PrivKey, []peer.ID, []*Node) {
+	cdns := newCachedDNS(dnsCacheRefreshInterval)
+	defer cdns.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	seed, err := newSeed()
+	require.NoError(t, err)
+
+	sks := make([]crypto.PrivKey, n)
+	pids := make([]peer.ID, n)
+
+	for i := 0; i < n; i++ {
+		sks[i], err = deriveKey(seed, deriveInfo(i))
+		require.NoError(t, err)
+
+		pids[i], err = peer.IDFromPrivateKey(sks[i])
+		require.NoError(t, err)
+	}
+
+	nodes := make([]*Node, n)
+
+	for i := 0; i < n; i++ {
+		cfg := Config{
+			DataDir:        t.TempDir(),
+			BlockstoreType: "flatfs",
+			AcceleratedDHT: false,
+			DHTSharedHost:  true,
+		}
+
+		// Add all remaining peers to the peering config.
+		for j, pid := range pids {
+			if j == i {
+				continue
+			}
+			cfg.Peering = append(cfg.Peering, peer.AddrInfo{ID: pid})
+		}
+
+		nodes[i], err = Setup(ctx, cfg, sks[i], cdns)
+		require.NoError(t, err)
+	}
+
+	require.Eventually(t, func() bool {
+		for _, node := range nodes {
+			peering := node.ps.ListPeers()
+			if len(peering) != n-1 {
+				return false
+			}
+
+			for _, peer := range peering {
+				peerInfo := node.host.Peerstore().PeerInfo(peer.ID)
+				if len(peerInfo.Addrs) == 0 {
+					return false
+				}
+			}
+		}
+
+		return true
+	}, time.Minute, time.Second)
+	return sks, pids, nodes
+}
+
+func TestSeedPeering(t *testing.T) {
+	testSeedPeering(t, 3)
+}