diff --git a/tests/utils.go b/tests/utils.go index 7d660f8e..eb839668 100644 --- a/tests/utils.go +++ b/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 } diff --git a/waku/v2/protocol/relay/waku_relay.go b/waku/v2/protocol/relay/waku_relay.go index 0c65dbe3..77d5a579 100644 --- a/waku/v2/protocol/relay/waku_relay.go +++ b/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 - opts []pubsub.Option - pubsub *pubsub.PubSub - params pubsub.GossipSubParams - timesource timesource.Timesource + host host.Host + opts []pubsub.Option + pubsub *pubsub.PubSub + params pubsub.GossipSubParams + 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 - IPColocationFactorWeight: -50, // p6: penalizes peers sharing more than threshold ips - IPColocationFactorThreshold: 5.0, // - BehaviourPenaltyWeight: -10, // p7: penalizes bad behaviour (weight and decay) - BehaviourPenaltyDecay: 0.986, + AppSpecificWeight: 0.0, + // p6: penalizes peers sharing more than threshold ips + IPColocationFactorWeight: -50, + IPColocationFactorThreshold: 5.0, + // 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 } diff --git a/waku/v2/protocol/relay/waku_relay_test.go b/waku/v2/protocol/relay/waku_relay_test.go index 9b68507b..22bbc167 100644 --- a/waku/v2/protocol/relay/waku_relay_test.go +++ b/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} diff --git a/waku/v2/rpc/admin_test.go b/waku/v2/rpc/admin_test.go index 513d1bbf..c23d78f9 100644 --- a/waku/v2/rpc/admin_test.go +++ b/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) }