[Networking] GossipSub iWant Flooding Mitigation

config/default-config.yml

@@ -106,6 +106,12 @@ network-config:
   ihave-async-inspection-sample-size-percentage: .10
   # Max number of ihave messages in a sample to be inspected
   ihave-max-sample-size: 100
+
+  # Max number of iwant messages in a sample to be inspected
+  gossipsub-rpc-iwant-max-sample-size: 1_000_000
+  # The allowed threshold of iWant messages receievd without a corresponding tracked iHave message that was sent
+  gossipsub-rpc-iwant-cache-miss-threshold: .5
+
   # RPC metrics observer inspector configs
   # The number of metrics inspector pool workers
   gossipsub-rpc-metrics-inspector-workers: 1

insecure/integration/functional/test/gossipsub/rpc_inspector/validation_inspector_test.go

@@ -608,7 +608,7 @@ func TestValidationInspector_DuplicateTopicId_Detection(t *testing.T) {
 			notification, ok := args[0].(*p2p.InvCtrlMsgNotif)
 			require.True(t, ok)
 			require.Equal(t, spammer.SpammerNode.Host().ID(), notification.PeerID)
-			require.True(t, validation.IsErrDuplicateTopic(notification.Err))
+			require.True(t, validation.IsDuplicateFoundErr(notification.Err))
 			require.Equal(t, messageCount, notification.Count)
 			switch notification.MsgType {
 			case p2pmsg.CtrlMsgGraft:

network/netconf/flags.go

@@ -59,6 +59,9 @@
 	metricsInspectorNumberOfWorkers = "gossipsub-rpc-metrics-inspector-workers"
 	metricsInspectorCacheSize       = "gossipsub-rpc-metrics-inspector-cache-size"
 
+	iwantMaxSampleSize      = "gossipsub-rpc-iwant-max-sample-size"
+	iwantCacheMissThreshold = "gossipsub-rpc-iwant-cache-miss-threshold"
+
 	alspDisabled            = "alsp-disable-penalty"
 	alspSpamRecordCacheSize = "alsp-spam-record-cache-size"
 	alspSpamRecordQueueSize = "alsp-spam-report-queue-size"
@@ -73,6 +76,7 @@
 		scoreTracerInterval, gossipSubRPCInspectorNotificationCacheSize, validationInspectorNumberOfWorkers, validationInspectorInspectMessageQueueCacheSize, validationInspectorClusterPrefixedTopicsReceivedCacheSize,
 		validationInspectorClusterPrefixedTopicsReceivedCacheDecay, validationInspectorClusterPrefixHardThreshold, ihaveSyncSampleSizePercentage, ihaveAsyncSampleSizePercentage,
 		ihaveMaxSampleSize, metricsInspectorNumberOfWorkers, metricsInspectorCacheSize, alspDisabled, alspSpamRecordCacheSize, alspSpamRecordQueueSize, alspHearBeatInterval,
+		iwantMaxSampleSize, iwantCacheMissThreshold,
 	}
 }
 
@@ -133,6 +137,9 @@
 	flags.Float64(ihaveSyncSampleSizePercentage, config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.IHaveSyncInspectSampleSizePercentage, "percentage of ihave messages to sample during synchronous validation")
 	flags.Float64(ihaveAsyncSampleSizePercentage, config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.IHaveAsyncInspectSampleSizePercentage, "percentage of ihave messages to sample during asynchronous validation")
 	flags.Float64(ihaveMaxSampleSize, config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.IHaveInspectionMaxSampleSize, "max number of ihaves to sample when performing validation")
+
+	flags.Uint64(iwantMaxSampleSize, config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.IWantRPCInspectionConfig.MaxSampleSize, "max number of ihaves to sample when performing validation")
+	flags.Float64(iwantCacheMissThreshold, config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.IWantRPCInspectionConfig.CacheMissThreshold, "max number of ihaves to sample when performing validation")
 }
 
 // rpcInspectorValidationLimits utility func that adds flags for each of the validation limits for each control message type.

network/p2p/inspector/validation/control_message_validation_inspector.go

@@ -54,6 +54,7 @@ type ControlMsgValidationInspector struct {
 	tracker      *cache.ClusterPrefixedMessagesReceivedTracker
 	idProvider   module.IdentityProvider
 	rateLimiters map[p2pmsg.ControlMessageType]p2p.BasicRateLimiter
+	rpcTracker   p2p.RPCControlTracking
 }
 
 var _ component.Component = (*ControlMsgValidationInspector)(nil)
@@ -196,6 +197,57 @@ func (c *ControlMsgValidationInspector) Inspect(from peer.ID, rpc *pubsub.RPC) e
 	return nil
 }
 
+// inspectIWant inspects RPC iWant control messages. This func will sample the iWants and perform validation on each iWant in the sample.
+// Ensuring that the following are true:
+// - Each iWant corresponds to an iHave that was sent.
+// - Each topic in the iWant sample is a valid topic.
+// If the number of iWants that do not have a corresponding iHave exceed the configured threshold an error is returned.
+// Args:
+// - iWant: the list of iWant control messages.
+// Returns:
+// - DuplicateFoundErr: if there are any duplicate message ids found in across any of the iWants.
+// - IWantCacheMissThresholdErr: if the rate of cache misses exceeds the configured allowed threshold.
+// - error: if any error occurs while sampling the iWants
+func (c *ControlMsgValidationInspector) inspectIWant(iWants []*pubsub_pb.ControlIWant) error {
+	sampleSize := uint(10 * c.rpcTracker.LastHighestIHaveRPCSize())
+	if sampleSize == 0 || sampleSize > c.config.IWantRPCInspectionConfig.MaxSampleSize {
+		sampleSize = c.config.IWantRPCInspectionConfig.MaxSampleSize
+	}
+
+	swap := func(i, j uint) {
+		iWants[i], iWants[j] = iWants[j], iWants[i]
+	}
+	err := c.sampleCtrlMessages(uint(len(iWants)), sampleSize, swap)
+	if err != nil {
+		return fmt.Errorf("failed to sample iwant messages: %w", err)
+	}
+
+	tracker := make(duplicateStrTracker)
+	cacheMisses := float64(0)
+	totalMessageIDS := float64(0)
+	for i := uint(0); i < sampleSize; i++ {
+		iWant := iWants[i]
+		for _, messageID := range iWant.GetMessageIDs() {
+			if tracker.isDuplicate(messageID) {
+				return NewDuplicateFoundErr(fmt.Errorf("duplicate message ID found: %s", messageID))
+			}
+
+			if !c.rpcTracker.WasIHaveRPCSent(messageID) {
+				cacheMisses++
+			}
+			tracker.set(messageID)
+			totalMessageIDS++
+		}
+	}
+
+	// check cache miss rate
+	if cacheMisses/totalMessageIDS > c.config.IWantRPCInspectionConfig.CacheMissThreshold {
+		return NewIWantCacheMissThresholdErr(cacheMisses, totalMessageIDS, c.config.IWantRPCInspectionConfig.CacheMissThreshold)
+	}
+
+	return nil
+}
+
 // Name returns the name of the rpc inspector.
 func (c *ControlMsgValidationInspector) Name() string {
 	return rpcInspectorComponentName
@@ -267,9 +319,16 @@ func (c *ControlMsgValidationInspector) blockingIHaveSamplePreprocessing(from pe
 // If the RPC control message count exceeds the configured hard threshold we perform synchronous topic validation on a subset
 // of the control messages. This is used for control message types that do not have an upper bound on the amount of messages a node can send.
 func (c *ControlMsgValidationInspector) blockingPreprocessingSampleRpc(from peer.ID, validationConfig *p2pconf.CtrlMsgValidationConfig, controlMessage *pubsub_pb.ControlMessage, sampleSize uint) error {
-	if validationConfig.ControlMsg != p2pmsg.CtrlMsgIHave && validationConfig.ControlMsg != p2pmsg.CtrlMsgIWant {
-		return fmt.Errorf("unexpected control message type %s encountered during blocking pre-processing sample rpc, expected %s or %s", validationConfig.ControlMsg, p2pmsg.CtrlMsgIHave, p2pmsg.CtrlMsgIWant)
+	if validationConfig.ControlMsg != p2pmsg.CtrlMsgIHave {
+		return fmt.Errorf("unexpected control message type %s encountered during blocking pre-processing sample rpc, expected %s", validationConfig.ControlMsg, p2pmsg.CtrlMsgIHave)
 	}
+
+	iHaves := controlMessage.GetIhave()
+	totalIhaves := uint(len(iHaves))
+	swap := func(i, j uint) {
+		iHaves[i], iHaves[j] = iHaves[j], iHaves[i]
+	}
+
 	activeClusterIDS := c.tracker.GetActiveClusterIds()
 	count := c.getCtrlMsgCount(validationConfig.ControlMsg, controlMessage)
 	lg := c.logger.With().
@@ -280,7 +339,7 @@ func (c *ControlMsgValidationInspector) blockingPreprocessingSampleRpc(from peer
 	if count > validationConfig.HardThreshold {
 		// for iHave control message topic validation we only validate a random subset of the messages
 		// shuffle the ihave messages to perform random validation on a subset of size sampleSize
-		err := c.sampleCtrlMessages(p2pmsg.CtrlMsgIHave, controlMessage, sampleSize)
+		err := c.sampleCtrlMessages(totalIhaves, sampleSize, swap)
 		if err != nil {
 			return fmt.Errorf("failed to sample ihave messages: %w", err)
 		}
@@ -306,7 +365,7 @@ func (c *ControlMsgValidationInspector) blockingPreprocessingSampleRpc(from peer
 	// to randomize async validation to avoid data race that can occur when
 	// performing the sampling asynchronously.
 	// for iHave control message topic validation we only validate a random subset of the messages
-	err := c.sampleCtrlMessages(p2pmsg.CtrlMsgIHave, controlMessage, sampleSize)
+	err := c.sampleCtrlMessages(totalIhaves, sampleSize, swap)
 	if err != nil {
 		return fmt.Errorf("failed to sample ihave messages: %w", err)
 	}
@@ -315,17 +374,10 @@ func (c *ControlMsgValidationInspector) blockingPreprocessingSampleRpc(from peer
 
 // sampleCtrlMessages performs sampling on the specified control message that will randomize
 // the items in the control message slice up to index sampleSize-1.
-func (c *ControlMsgValidationInspector) sampleCtrlMessages(ctrlMsgType p2pmsg.ControlMessageType, ctrlMsg *pubsub_pb.ControlMessage, sampleSize uint) error {
-	switch ctrlMsgType {
-	case p2pmsg.CtrlMsgIHave:
-		iHaves := ctrlMsg.GetIhave()
-		swap := func(i, j uint) {
-			iHaves[i], iHaves[j] = iHaves[j], iHaves[i]
-		}
-		err := flowrand.Samples(uint(len(iHaves)), sampleSize, swap)
-		if err != nil {
-			return fmt.Errorf("failed to get random sample of ihave control messages: %w", err)
-		}
+func (c *ControlMsgValidationInspector) sampleCtrlMessages(totalSize, sampleSize uint, swap func(i, j uint)) error {
+	err := flowrand.Samples(totalSize, sampleSize, swap)
+	if err != nil {
+		return fmt.Errorf("failed to get random sample of ihave control messages: %w", err)
 	}
 	return nil
 }
@@ -339,6 +391,12 @@ func (c *ControlMsgValidationInspector) processInspectMsgReq(req *InspectMsgRequ
 		c.metrics.AsyncProcessingFinished(req.validationConfig.ControlMsg.String(), time.Since(start))
 	}()
 
+	// iWant validation uses new sample size validation. This will be updated for all other control message types.
+	switch req.validationConfig.ControlMsg {
+	case p2pmsg.CtrlMsgIWant:
+		return c.inspectIWant(req.ctrlMsg.GetIwant())
+	}
+
 	count := c.getCtrlMsgCount(req.validationConfig.ControlMsg, req.ctrlMsg)
 	lg := c.logger.With().
 		Str("peer_id", req.Peer.String()).
@@ -392,7 +450,7 @@ func (c *ControlMsgValidationInspector) getCtrlMsgCount(ctrlMsgType p2pmsg.Contr
 // validateTopics ensures all topics in the specified control message are valid flow topic/channel and no duplicate topics exist.
 // Expected error returns during normal operations:
 //   - channels.InvalidTopicErr: if topic is invalid.
-//   - ErrDuplicateTopic: if a duplicate topic ID is encountered.
+//   - DuplicateFoundErr: if a duplicate topic ID is encountered.
 func (c *ControlMsgValidationInspector) validateTopics(from peer.ID, validationConfig *p2pconf.CtrlMsgValidationConfig, ctrlMsg *pubsub_pb.ControlMessage) error {
 	activeClusterIDS := c.tracker.GetActiveClusterIds()
 	switch validationConfig.ControlMsg {
@@ -413,13 +471,13 @@ func (c *ControlMsgValidationInspector) validateTopics(from peer.ID, validationC
 
 // validateGrafts performs topic validation on all grafts in the control message using the provided validateTopic func while tracking duplicates.
 func (c *ControlMsgValidationInspector) validateGrafts(from peer.ID, ctrlMsg *pubsub_pb.ControlMessage, activeClusterIDS flow.ChainIDList) error {
-	tracker := make(duplicateTopicTracker)
+	tracker := make(duplicateStrTracker)
 	for _, graft := range ctrlMsg.GetGraft() {
 		topic := channels.Topic(graft.GetTopicID())
-		if tracker.isDuplicate(topic) {
-			return NewDuplicateTopicErr(topic)
+		if tracker.isDuplicate(topic.String()) {
+			return NewDuplicateFoundErr(fmt.Errorf("duplicate topic found: %s", topic.String()))
 		}
-		tracker.set(topic)
+		tracker.set(topic.String())
 		err := c.validateTopic(from, topic, activeClusterIDS)
 		if err != nil {
 			return err
@@ -430,13 +488,13 @@ func (c *ControlMsgValidationInspector) validateGrafts(from peer.ID, ctrlMsg *pu
 
 // validatePrunes performs topic validation on all prunes in the control message using the provided validateTopic func while tracking duplicates.
 func (c *ControlMsgValidationInspector) validatePrunes(from peer.ID, ctrlMsg *pubsub_pb.ControlMessage, activeClusterIDS flow.ChainIDList) error {
-	tracker := make(duplicateTopicTracker)
+	tracker := make(duplicateStrTracker)
 	for _, prune := range ctrlMsg.GetPrune() {
 		topic := channels.Topic(prune.GetTopicID())
-		if tracker.isDuplicate(topic) {
-			return NewDuplicateTopicErr(topic)
+		if tracker.isDuplicate(topic.String()) {
+			return NewDuplicateFoundErr(fmt.Errorf("duplicate topic found: %s", topic.String()))
 		}
-		tracker.set(topic)
+		tracker.set(topic.String())
 		err := c.validateTopic(from, topic, activeClusterIDS)
 		if err != nil {
 			return err
@@ -455,15 +513,15 @@ func (c *ControlMsgValidationInspector) validateIhaves(from peer.ID, validationC
 // Sample size ensures liveness of the network when validating messages with no upper bound on the amount of messages that may be received.
 // All errors returned from this function can be considered benign.
 func (c *ControlMsgValidationInspector) validateTopicsSample(from peer.ID, validationConfig *p2pconf.CtrlMsgValidationConfig, ctrlMsg *pubsub_pb.ControlMessage, activeClusterIDS flow.ChainIDList, sampleSize uint) error {
-	tracker := make(duplicateTopicTracker)
+	tracker := make(duplicateStrTracker)
 	switch validationConfig.ControlMsg {
 	case p2pmsg.CtrlMsgIHave:
 		for i := uint(0); i < sampleSize; i++ {
 			topic := channels.Topic(ctrlMsg.Ihave[i].GetTopicID())
-			if tracker.isDuplicate(topic) {
-				return NewDuplicateTopicErr(topic)
+			if tracker.isDuplicate(topic.String()) {
+				return NewDuplicateFoundErr(fmt.Errorf("duplicate topic found: %s", topic.String()))
 			}
-			tracker.set(topic)
+			tracker.set(topic.String())
 			err := c.validateTopic(from, topic, activeClusterIDS)
 			if err != nil {
 				return err

network/p2p/inspector/validation/errors.go

@@ -8,6 +8,28 @@ import (
 	p2pmsg "github.com/onflow/flow-go/network/p2p/message"
 )
 
+// IWantCacheMissThresholdErr indicates that the amount of cache misses exceeds the allowed threshold.
+type IWantCacheMissThresholdErr struct {
+	misses          float64
+	totalMessageIDS float64
+	threshold       float64
+}
+
+func (e IWantCacheMissThresholdErr) Error() string {
+	return fmt.Sprintf("%f/%f iWant cache misses exceeds the allowed threshold: %f", e.misses, e.totalMessageIDS, e.threshold)
+}
+
+// NewIWantCacheMissThresholdErr returns a new IWantCacheMissThresholdErr.
+func NewIWantCacheMissThresholdErr(misses, totalMessageIDS, threshold float64) IWantCacheMissThresholdErr {
+	return IWantCacheMissThresholdErr{misses, totalMessageIDS, threshold}
+}
+
+// IsIWantCacheMissThresholdErr returns true if an error is IWantCacheMissThresholdErr
+func IsIWantCacheMissThresholdErr(err error) bool {
+	var e IWantCacheMissThresholdErr
+	return errors.As(err, &e)
+}
+
 // ErrHardThreshold indicates that the amount of RPC messages received exceeds hard threshold.
 type ErrHardThreshold struct {
 	// controlMsg the control message type.
@@ -53,23 +75,23 @@ func IsErrRateLimitedControlMsg(err error) bool {
 	return errors.As(err, &e)
 }
 
-// ErrDuplicateTopic error that indicates a duplicate topic in control message has been detected.
-type ErrDuplicateTopic struct {
-	topic channels.Topic
+// DuplicateFoundErr error that indicates a duplicate has been detected. This can be duplicate topic or message ID tracking.
+type DuplicateFoundErr struct {
+	err error
 }
 
-func (e ErrDuplicateTopic) Error() string {
-	return fmt.Errorf("duplicate topic %s", e.topic).Error()
+func (e DuplicateFoundErr) Error() string {
+	return e.err.Error()
 }
 
-// NewDuplicateTopicErr returns a new ErrDuplicateTopic.
-func NewDuplicateTopicErr(topic channels.Topic) ErrDuplicateTopic {
-	return ErrDuplicateTopic{topic: topic}
+// NewDuplicateFoundErr returns a new DuplicateFoundErr.
+func NewDuplicateFoundErr(err error) DuplicateFoundErr {
+	return DuplicateFoundErr{err}
 }
 
-// IsErrDuplicateTopic returns true if an error is ErrDuplicateTopic.
-func IsErrDuplicateTopic(err error) bool {
-	var e ErrDuplicateTopic
+// IsDuplicateFoundErr returns true if an error is DuplicateFoundErr.
+func IsDuplicateFoundErr(err error) bool {
+	var e DuplicateFoundErr
 	return errors.As(err, &e)
 }
 

network/p2p/inspector/validation/errors_test.go

@@ -63,19 +63,20 @@ func TestErrRateLimitedControlMsgRoundTrip(t *testing.T) {
 	assert.False(t, IsErrRateLimitedControlMsg(dummyErr), "IsErrRateLimitedControlMsg should return false for non-ErrRateLimitedControlMsg error")
 }
 
-// TestErrDuplicateTopicRoundTrip ensures correct error formatting for ErrDuplicateTopic.
+// TestErrDuplicateTopicRoundTrip ensures correct error formatting for DuplicateFoundErr.
 func TestErrDuplicateTopicRoundTrip(t *testing.T) {
 	topic := channels.Topic("topic")
-	err := NewDuplicateTopicErr(topic)
+	e := fmt.Errorf("duplicate topic found: %s", topic.String())
+	err := NewDuplicateFoundErr(e)
 
 	// tests the error message formatting.
-	expectedErrMsg := fmt.Errorf("duplicate topic %s", topic).Error()
+	expectedErrMsg := e.Error()
 	assert.Equal(t, expectedErrMsg, err.Error(), "the error message should be correctly formatted")
 
-	// tests the IsErrDuplicateTopic function.
-	assert.True(t, IsErrDuplicateTopic(err), "IsErrDuplicateTopic should return true for ErrDuplicateTopic error")
+	// tests the IsDuplicateFoundErr function.
+	assert.True(t, IsDuplicateFoundErr(err), "IsDuplicateFoundErr should return true for DuplicateFoundErr error")
 
-	// test IsErrDuplicateTopic with a different error type.
+	// test IsDuplicateFoundErr with a different error type.
 	dummyErr := fmt.Errorf("dummy error")
-	assert.False(t, IsErrDuplicateTopic(dummyErr), "IsErrDuplicateTopic should return false for non-ErrDuplicateTopic error")
+	assert.False(t, IsDuplicateFoundErr(dummyErr), "IsDuplicateFoundErr should return false for non-DuplicateFoundErr error")
 }

network/p2p/inspector/validation/utils.go

@@ -1,14 +1,12 @@
 package validation
 
-import "github.com/onflow/flow-go/network/channels"
+type duplicateStrTracker map[string]struct{}
 
-type duplicateTopicTracker map[channels.Topic]struct{}
-
-func (d duplicateTopicTracker) set(topic channels.Topic) {
-	d[topic] = struct{}{}
+func (d duplicateStrTracker) set(s string) {
+	d[s] = struct{}{}
 }
 
-func (d duplicateTopicTracker) isDuplicate(topic channels.Topic) bool {
-	_, ok := d[topic]
+func (d duplicateStrTracker) isDuplicate(s string) bool {
+	_, ok := d[s]
 	return ok
 }