fix: code review and disconnect bad peers

tests/utils.go

@@ -97,7 +97,7 @@ func MakeHost(ctx context.Context, port int, randomness io.Reader) (host.Host, e
 	}
 
 	// 0.0.0.0 will listen on any interface device.
-	sourceMultiAddr, err := multiaddr.NewMultiaddr(fmt.Sprintf("/ip4/0.0.0.0/tcp/%d", port))
+	sourceMultiAddr, err := multiaddr.NewMultiaddr(fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", port))
 	if err != nil {
 		return nil, err
 	}

waku/v2/protocol/relay/waku_relay.go

@@ -30,11 +30,14 @@ const WakuRelayID_v200 = protocol.ID("/vac/waku/relay/2.0.0")
 var DefaultWakuTopic string = waku_proto.DefaultPubsubTopic().String()
 
 type WakuRelay struct {
-	host       host.Host
+	host                host.Host
-	opts       []pubsub.Option
+	opts                []pubsub.Option
-	pubsub     *pubsub.PubSub
+	pubsub              *pubsub.PubSub
-	params     pubsub.GossipSubParams
+	params              pubsub.GossipSubParams
-	timesource timesource.Timesource
+	peerScoreParams     *pubsub.PeerScoreParams
+	peerScoreThresholds *pubsub.PeerScoreThresholds
+	topicParams         *pubsub.TopicScoreParams
+	timesource          timesource.Timesource
 
 	log *zap.Logger
 
@@ -81,26 +84,54 @@ func NewWakuRelay(bcaster Broadcaster, minPeersToPublish int, timesource timesou
 	cfg.HistoryGossip = 3
 	cfg.FanoutTTL = time.Minute
 
-	peerScoreParams := &pubsub.PeerScoreParams{
+	w.peerScoreParams = &pubsub.PeerScoreParams{
+		Topics:        make(map[string]*pubsub.TopicScoreParams),
 		DecayInterval: 12 * time.Second, // how often peer scoring is updated
 		DecayToZero:   0.01,             // below this we consider the parameter to be zero
 		RetainScore:   10 * time.Minute, // remember peer score during x after it disconnects
+		// p5: application specific, unset
 		AppSpecificScore: func(p peer.ID) float64 {
 			return 0
 		},
-		AppSpecificWeight:           0.0, // p5: application specific, unset
+		AppSpecificWeight: 0.0,
-		IPColocationFactorWeight:    -50, // p6: penalizes peers sharing more than threshold ips
+		// p6: penalizes peers sharing more than threshold ips
-		IPColocationFactorThreshold: 5.0, //
+		IPColocationFactorWeight:    -50,
-		BehaviourPenaltyWeight:      -10, // p7: penalizes bad behaviour (weight and decay)
+		IPColocationFactorThreshold: 5.0,
-		BehaviourPenaltyDecay:       0.986,
+		// p7: penalizes bad behaviour (weight and decay)
+		BehaviourPenaltyWeight: -10,
+		BehaviourPenaltyDecay:  0.986,
 	}
 
-	peerScoreThresholds := &pubsub.PeerScoreThresholds{
+	w.peerScoreThresholds = &pubsub.PeerScoreThresholds{
 		GossipThreshold:             -100,   // no gossip is sent to peers below this score
 		PublishThreshold:            -1000,  // no self-published msgs are sent to peers below this score
 		GraylistThreshold:           -10000, // used to trigger disconnections + ignore peer if below this score
 		OpportunisticGraftThreshold: 0,      // grafts better peers if the mesh median score drops below this. unset.
+	}
 
+	w.topicParams = &pubsub.TopicScoreParams{
+		TopicWeight: 1,
+		// p1: favours peers already in the mesh
+		TimeInMeshWeight:  0.01,
+		TimeInMeshQuantum: time.Second,
+		TimeInMeshCap:     10.0,
+		// p2: rewards fast peers
+		FirstMessageDeliveriesWeight: 1.0,
+		FirstMessageDeliveriesDecay:  0.5,
+		FirstMessageDeliveriesCap:    10.0,
+		// p3: penalizes lazy peers. safe low value
+		MeshMessageDeliveriesWeight:     0,
+		MeshMessageDeliveriesDecay:      0,
+		MeshMessageDeliveriesCap:        0,
+		MeshMessageDeliveriesThreshold:  0,
+		MeshMessageDeliveriesWindow:     0,
+		MeshMessageDeliveriesActivation: 0,
+		// p3b: tracks history of prunes
+		MeshFailurePenaltyWeight: 0,
+		MeshFailurePenaltyDecay:  0,
+		// p4: penalizes invalid messages. highly penalize peers sending wrong messages
+		InvalidMessageDeliveriesWeight: -100.0,
+		InvalidMessageDeliveriesDecay:  0.5,
 	}
 
 	// default options required by WakuRelay
@@ -124,7 +155,8 @@ func NewWakuRelay(bcaster Broadcaster, minPeersToPublish int, timesource timesou
 		pubsub.WithGossipSubParams(cfg),
 		pubsub.WithFloodPublish(true),
 		pubsub.WithSeenMessagesTTL(2 * time.Minute),
-		pubsub.WithPeerScore(peerScoreParams, peerScoreThresholds),
+		pubsub.WithPeerScore(w.peerScoreParams, w.peerScoreThresholds),
+		pubsub.WithPeerScoreInspect(w.peerScoreInspector, 6*time.Second),
 		pubsub.WithDefaultValidator(func(ctx context.Context, peerID peer.ID, message *pubsub.Message) bool {
 			msg := new(pb.WakuMessage)
 			err := proto.Unmarshal(message.Data, msg)
@@ -135,6 +167,22 @@ func NewWakuRelay(bcaster Broadcaster, minPeersToPublish int, timesource timesou
 	return w
 }
 
+func (w *WakuRelay) peerScoreInspector(peerScoresSnapshots map[peer.ID]*pubsub.PeerScoreSnapshot) {
+	if w.host == nil {
+		return
+	}
+
+	for pid, snap := range peerScoresSnapshots {
+		if snap.Score < w.peerScoreThresholds.GraylistThreshold {
+			// Disconnect bad peers
+			err := w.host.Network().ClosePeer(pid)
+			if err != nil {
+				w.log.Error("could not disconnect peer", logging.HostID("peer", pid), zap.Error(err))
+			}
+		}
+	}
+}
+
 // Sets the host to be able to mount or consume a protocol
 func (w *WakuRelay) SetHost(h host.Host) {
 	w.host = h
@@ -195,6 +243,12 @@ func (w *WakuRelay) upsertTopic(topic string) (*pubsub.Topic, error) {
 		if err != nil {
 			return nil, err
 		}
+
+		err = newTopic.SetScoreParams(w.topicParams)
+		if err != nil {
+			return nil, err
+		}
+
 		w.wakuRelayTopics[topic] = newTopic
 		pubSubTopic = newTopic
 	}

waku/v2/protocol/relay/waku_relay_test.go

@@ -3,9 +3,13 @@ package relay
 import (
 	"context"
 	"crypto/rand"
+	"fmt"
 	"testing"
+	"time"
 
 	pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb"
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/peerstore"
 	"github.com/stretchr/testify/require"
 	"github.com/waku-org/go-waku/tests"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
@@ -56,6 +60,83 @@ func TestWakuRelay(t *testing.T) {
 	<-ctx.Done()
 }
 
+func createRelayNode(t *testing.T) (host.Host, *WakuRelay) {
+	port, err := tests.FindFreePort(t, "", 5)
+	require.NoError(t, err)
+	host, err := tests.MakeHost(context.Background(), port, rand.Reader)
+	require.NoError(t, err)
+
+	relay := NewWakuRelay(nil, 0, timesource.NewDefaultClock(), utils.Logger())
+	relay.SetHost(host)
+	return host, relay
+}
+
+func TestGossipsubScore(t *testing.T) {
+	testTopic := "/waku/2/go/relay/test"
+
+	hosts := make([]host.Host, 5)
+	relay := make([]*WakuRelay, 5)
+	for i := 0; i < 5; i++ {
+		hosts[i], relay[i] = createRelayNode(t)
+		if i == 0 {
+			// This is a hack to remove the default validator from the list of default options
+			relay[i].opts = relay[i].opts[:len(relay[i].opts)-1]
+		}
+		err := relay[i].Start(context.Background())
+		require.NoError(t, err)
+	}
+
+	for i := 0; i < 5; i++ {
+		for j := 0; j < 5; j++ {
+			if i == j {
+				continue
+			}
+
+			hosts[i].Peerstore().AddAddrs(hosts[j].ID(), hosts[j].Addrs(), peerstore.PermanentAddrTTL)
+			err := hosts[i].Connect(context.Background(), hosts[j].Peerstore().PeerInfo(hosts[j].ID()))
+			require.NoError(t, err)
+		}
+
+		sub, err := relay[i].subscribe(testTopic)
+		require.NoError(t, err)
+		go func() {
+			for {
+				_, err := sub.Next(context.Background())
+				if err != nil {
+					fmt.Println(err)
+				}
+			}
+		}()
+	}
+
+	time.Sleep(2 * time.Second)
+
+	for i := 0; i < 5; i++ {
+		require.Len(t, hosts[i].Network().Conns(), 4)
+	}
+
+	// We obtain the go-libp2p topic directly because we normally can't publish anything other than WakuMessages
+	pubsubTopic, err := relay[0].upsertTopic(testTopic)
+	require.NoError(t, err)
+	for i := 0; i < 50; i++ {
+		buf := make([]byte, 1000)
+		_, err := rand.Read(buf)
+		require.NoError(t, err)
+		err = pubsubTopic.Publish(context.Background(), buf)
+		require.NoError(t, err)
+	}
+
+	// long wait, must be higher than the configured decayInterval (how often score is updated)
+	time.Sleep(20 * time.Second)
+
+	// nodes[0] was blacklisted from all other peers, no connections
+	require.Len(t, hosts[0].Network().Conns(), 0)
+
+	for i := 1; i < 5; i++ {
+		require.Len(t, hosts[i].Network().Conns(), 3)
+	}
+}
+
 func TestMsgID(t *testing.T) {
 	expectedMsgIdBytes := []byte{208, 214, 63, 55, 144, 6, 206, 39, 40, 251, 138, 74, 66, 168, 43, 32, 91, 94, 149, 122, 237, 198, 149, 87, 232, 156, 197, 34, 53, 131, 78, 112}
 

waku/v2/rpc/admin_test.go

@@ -69,5 +69,5 @@ func TestV1Peers(t *testing.T) {
 
 	err = a.GetV1Peers(request, &GetPeersArgs{}, &reply)
 	require.NoError(t, err)
-	require.Len(t, reply, 2)
+	require.Len(t, reply, 1)
 }