Merge pull request onflow#4978 from onflow/khalil/6664-invalid-contro…

…l-message-weight

[Networking] Apply GossipSub Spam Penalty to Misbehaving Peers Based on `Count` and `Err` in `InvCtrlMsgNotif`

cmd/access/node_builder/access_node_builder.go

@@ -1741,38 +1741,49 @@ func (builder *FlowAccessNodeBuilder) initPublicLibp2pNode(networkKey crypto.Pri
 		return nil, fmt.Errorf("could not create connection manager: %w", err)
 	}
 
-	libp2pNode, err := p2pbuilder.NewNodeBuilder(builder.Logger, &builder.FlowConfig.NetworkConfig.GossipSub, &p2pbuilderconfig.MetricsConfig{
-		HeroCacheFactory: builder.HeroCacheMetricsFactory(),
-		Metrics:          networkMetrics,
-	},
-		network.PublicNetwork,
-		bindAddress,
-		networkKey,
-		builder.SporkID,
-		builder.IdentityProvider,
-		&builder.FlowConfig.NetworkConfig.ResourceManager,
-		&p2pbuilderconfig.PeerManagerConfig{
+	params := &p2pbuilder.LibP2PNodeBuilderConfig{
+		Logger: builder.Logger,
+		MetricsConfig: &p2pbuilderconfig.MetricsConfig{
+			HeroCacheFactory: builder.HeroCacheMetricsFactory(),
+			Metrics:          networkMetrics,
+		},
+		NetworkingType:        network.PublicNetwork,
+		Address:               bindAddress,
+		NetworkKey:            networkKey,
+		SporkId:               builder.SporkID,
+		IdProvider:            builder.IdentityProvider,
+		ResourceManagerParams: &builder.FlowConfig.NetworkConfig.ResourceManager,
+		RpcInspectorParams:    &builder.FlowConfig.NetworkConfig.GossipSub.RpcInspector,
+		PeerManagerParams: &p2pbuilderconfig.PeerManagerConfig{
 			// TODO: eventually, we need pruning enabled even on public network. However, it needs a modified version of
 			// the peer manager that also operate on the public identities.
 			ConnectionPruning: connection.PruningDisabled,
 			UpdateInterval:    builder.FlowConfig.NetworkConfig.PeerUpdateInterval,
 			ConnectorFactory:  connection.DefaultLibp2pBackoffConnectorFactory(),
 		},
-		&p2p.DisallowListCacheConfig{
+		SubscriptionProviderParams: &builder.FlowConfig.NetworkConfig.GossipSub.SubscriptionProvider,
+		DisallowListCacheCfg: &p2p.DisallowListCacheConfig{
 			MaxSize: builder.FlowConfig.NetworkConfig.DisallowListNotificationCacheSize,
 			Metrics: metrics.DisallowListCacheMetricsFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
 		},
-		&p2pbuilderconfig.UnicastConfig{
-			Unicast: builder.FlowConfig.NetworkConfig.Unicast,
-		}).
+		UnicastConfig: &p2pbuilderconfig.UnicastConfig{
+			Unicast:                builder.FlowConfig.NetworkConfig.Unicast,
+			RateLimiterDistributor: builder.UnicastRateLimiterDistributor,
+		},
+		GossipSubCfg: &builder.FlowConfig.NetworkConfig.GossipSub,
+	}
+	nodeBuilder, err := p2pbuilder.NewNodeBuilder(params)
+	if err != nil {
+		return nil, fmt.Errorf("could not create libp2p node builder: %w", err)
+	}
+	libp2pNode, err := nodeBuilder.
 		SetBasicResolver(builder.Resolver).
 		SetSubscriptionFilter(subscription.NewRoleBasedFilter(flow.RoleAccess, builder.IdentityProvider)).
 		SetConnectionManager(connManager).
 		SetRoutingSystem(func(ctx context.Context, h host.Host) (routing.Routing, error) {
 			return dht.NewDHT(ctx, h, protocols.FlowPublicDHTProtocolID(builder.SporkID), builder.Logger, networkMetrics, dht.AsServer())
 		}).
 		Build()
-
 	if err != nil {
 		return nil, fmt.Errorf("could not build libp2p node for staked access node: %w", err)
 	}

cmd/observer/node_builder/observer_builder.go

@@ -742,27 +742,36 @@ func (builder *ObserverServiceBuilder) initPublicLibp2pNode(networkKey crypto.Pr
 		pis = append(pis, pi)
 	}
 
-	node, err := p2pbuilder.NewNodeBuilder(
-		builder.Logger,
-		&builder.FlowConfig.NetworkConfig.GossipSub,
-		&p2pbuilderconfig.MetricsConfig{
+	params := &p2pbuilder.LibP2PNodeBuilderConfig{
+		Logger: builder.Logger,
+		MetricsConfig: &p2pbuilderconfig.MetricsConfig{
 			HeroCacheFactory: builder.HeroCacheMetricsFactory(),
 			Metrics:          builder.Metrics.Network,
 		},
-		network.PublicNetwork,
-		builder.BaseConfig.BindAddr,
-		networkKey,
-		builder.SporkID,
-		builder.IdentityProvider,
-		&builder.FlowConfig.NetworkConfig.ResourceManager,
-		p2pbuilderconfig.PeerManagerDisableConfig(), // disable peer manager for observer node.
-		&p2p.DisallowListCacheConfig{
+		NetworkingType:             network.PublicNetwork,
+		Address:                    builder.BaseConfig.BindAddr,
+		NetworkKey:                 networkKey,
+		SporkId:                    builder.SporkID,
+		IdProvider:                 builder.IdentityProvider,
+		ResourceManagerParams:      &builder.FlowConfig.NetworkConfig.ResourceManager,
+		RpcInspectorParams:         &builder.FlowConfig.NetworkConfig.GossipSub.RpcInspector,
+		PeerManagerParams:          p2pbuilderconfig.PeerManagerDisableConfig(),
+		SubscriptionProviderParams: &builder.FlowConfig.NetworkConfig.GossipSub.SubscriptionProvider,
+		DisallowListCacheCfg: &p2p.DisallowListCacheConfig{
 			MaxSize: builder.FlowConfig.NetworkConfig.DisallowListNotificationCacheSize,
 			Metrics: metrics.DisallowListCacheMetricsFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
 		},
-		&p2pbuilderconfig.UnicastConfig{
-			Unicast: builder.FlowConfig.NetworkConfig.Unicast,
-		}).
+		UnicastConfig: &p2pbuilderconfig.UnicastConfig{
+			Unicast:                builder.FlowConfig.NetworkConfig.Unicast,
+			RateLimiterDistributor: builder.UnicastRateLimiterDistributor,
+		},
+		GossipSubCfg: &builder.FlowConfig.NetworkConfig.GossipSub,
+	}
+	nodeBuilder, err := p2pbuilder.NewNodeBuilder(params)
+	if err != nil {
+		return nil, fmt.Errorf("could not create libp2p node builder: %w", err)
+	}
+	libp2pNode, err := nodeBuilder.
 		SetSubscriptionFilter(
 			subscription.NewRoleBasedFilter(
 				subscription.UnstakedRole, builder.IdentityProvider,
@@ -782,7 +791,7 @@ func (builder *ObserverServiceBuilder) initPublicLibp2pNode(networkKey crypto.Pr
 		return nil, fmt.Errorf("could not initialize libp2p node for observer: %w", err)
 	}
 
-	builder.LibP2PNode = node
+	builder.LibP2PNode = libp2pNode
 
 	return builder.LibP2PNode, nil
 }

cmd/scaffold.go

@@ -380,28 +380,34 @@ func (fnb *FlowNodeBuilder) EnqueueNetworkInit() {
 		if err != nil {
 			return nil, fmt.Errorf("could not determine dht activation status: %w", err)
 		}
-		builder, err := p2pbuilder.DefaultNodeBuilder(fnb.Logger,
-			myAddr,
-			network.PrivateNetwork,
-			fnb.NetworkKey,
-			fnb.SporkID,
-			fnb.IdentityProvider,
-			&p2pbuilderconfig.MetricsConfig{
+
+		params := &p2pbuilder.LibP2PNodeBuilderConfig{
+			Logger: fnb.Logger,
+			MetricsConfig: &p2pbuilderconfig.MetricsConfig{
 				Metrics:          fnb.Metrics.Network,
 				HeroCacheFactory: fnb.HeroCacheMetricsFactory(),
 			},
+			NetworkingType:             network.PrivateNetwork,
+			Address:                    myAddr,
+			NetworkKey:                 fnb.NetworkKey,
+			SporkId:                    fnb.SporkID,
+			IdProvider:                 fnb.IdentityProvider,
+			ResourceManagerParams:      &fnb.FlowConfig.NetworkConfig.ResourceManager,
+			RpcInspectorParams:         &fnb.FlowConfig.NetworkConfig.GossipSub.RpcInspector,
+			PeerManagerParams:          peerManagerCfg,
+			SubscriptionProviderParams: &fnb.FlowConfig.NetworkConfig.GossipSub.SubscriptionProvider,
+			DisallowListCacheCfg: &p2p.DisallowListCacheConfig{
+				MaxSize: fnb.FlowConfig.NetworkConfig.DisallowListNotificationCacheSize,
+				Metrics: metrics.DisallowListCacheMetricsFactory(fnb.HeroCacheMetricsFactory(), network.PrivateNetwork),
+			},
+			UnicastConfig: uniCfg,
+			GossipSubCfg:  &fnb.FlowConfig.NetworkConfig.GossipSub,
+		}
+		builder, err := p2pbuilder.DefaultNodeBuilder(params,
 			fnb.Resolver,
 			fnb.BaseConfig.NodeRole,
 			connGaterCfg,
-			peerManagerCfg,
-			&fnb.FlowConfig.NetworkConfig.GossipSub,
-			&fnb.FlowConfig.NetworkConfig.ResourceManager,
-			uniCfg,
 			&fnb.FlowConfig.NetworkConfig.ConnectionManager,
-			&p2p.DisallowListCacheConfig{
-				MaxSize: fnb.FlowConfig.NetworkConfig.DisallowListNotificationCacheSize,
-				Metrics: metrics.DisallowListCacheMetricsFactory(fnb.HeroCacheMetricsFactory(), network.PrivateNetwork),
-			},
 			dhtActivationStatus)
 		if err != nil {
 			return nil, fmt.Errorf("could not create libp2p node builder: %w", err)

config/config_test.go

@@ -23,14 +23,31 @@ func TestBindPFlags(t *testing.T) {
 	t.Run("should override config values when any flag is set", func(t *testing.T) {
 		c := defaultConfig(t)
 		flags := testFlagSet(c)
+
 		err := flags.Set("networking-connection-pruning", "false")
 		require.NoError(t, err)
+		err = flags.Set("gossipsub-scoring-parameters-scoring-registry-misbehaviour-penalties-graft", "-10")
+		require.NoError(t, err)
+		err = flags.Set("gossipsub-scoring-parameters-scoring-registry-misbehaviour-penalties-prune", "-5")
+		require.NoError(t, err)
+		err = flags.Set("gossipsub-scoring-parameters-scoring-registry-misbehaviour-penalties-ihave", "-2")
+		require.NoError(t, err)
+		err = flags.Set("gossipsub-scoring-parameters-scoring-registry-misbehaviour-penalties-iwant", "-.9")
+		require.NoError(t, err)
+		err = flags.Set("gossipsub-scoring-parameters-scoring-registry-misbehaviour-penalties-publish", "-.1")
+		require.NoError(t, err)
+
 		require.NoError(t, flags.Parse(nil))
 
 		configFileUsed, err := BindPFlags(c, flags)
 		require.NoError(t, err)
 		require.False(t, configFileUsed)
 		require.False(t, c.NetworkConfig.NetworkConnectionPruning)
+		require.Equal(t, c.NetworkConfig.GossipSub.ScoringParameters.ScoringRegistryParameters.MisbehaviourPenalties.GraftMisbehaviour, float64(-10))
+		require.Equal(t, c.NetworkConfig.GossipSub.ScoringParameters.ScoringRegistryParameters.MisbehaviourPenalties.PruneMisbehaviour, float64(-5))
+		require.Equal(t, c.NetworkConfig.GossipSub.ScoringParameters.ScoringRegistryParameters.MisbehaviourPenalties.IHaveMisbehaviour, float64(-2))
+		require.Equal(t, c.NetworkConfig.GossipSub.ScoringParameters.ScoringRegistryParameters.MisbehaviourPenalties.IWantMisbehaviour, float64(-.9))
+		require.Equal(t, c.NetworkConfig.GossipSub.ScoringParameters.ScoringRegistryParameters.MisbehaviourPenalties.PublishMisbehaviour, float64(-.1))
 	})
 	t.Run("should return an error if flags are not parsed", func(t *testing.T) {
 		c := defaultConfig(t)

config/default-config.yml

@@ -512,15 +512,15 @@ network-config:
             skip-decay-threshold: -0.1
         misbehaviour-penalties:
           # The penalty applied to the application specific penalty when a peer conducts a graft misbehaviour.
-          graft: -10
+          graft: -0.01
           # The penalty applied to the application specific penalty when a peer conducts a prune misbehaviour.
-          prune: -10
+          prune: -0.01
           # The penalty applied to the application specific penalty when a peer conducts a iHave misbehaviour.
-          ihave: -10
+          ihave: -0.01
           # The penalty applied to the application specific penalty when a peer conducts a iWant misbehaviour.
-          iwant: -10
+          iwant: -0.01
           # The penalty applied to the application specific penalty when a peer conducts a rpc publish message misbehaviour.
-          publish: -10
+          publish: -0.01
           # The factor used to reduce the penalty for control message misbehaviours on cluster prefixed topics. This allows a more lenient punishment for nodes
           # that fall behind and may need to request old data.
           cluster-prefixed-reduction-factor: 0.2

follower/follower_builder.go

@@ -572,27 +572,36 @@ func (builder *FollowerServiceBuilder) initPublicLibp2pNode(networkKey crypto.Pr
 		pis = append(pis, pi)
 	}
 
-	node, err := p2pbuilder.NewNodeBuilder(
-		builder.Logger,
-		&builder.FlowConfig.NetworkConfig.GossipSub,
-		&p2pbuilderconfig.MetricsConfig{
+	params := &p2pbuilder.LibP2PNodeBuilderConfig{
+		Logger: builder.Logger,
+		MetricsConfig: &p2pbuilderconfig.MetricsConfig{
 			HeroCacheFactory: builder.HeroCacheMetricsFactory(),
 			Metrics:          builder.Metrics.Network,
 		},
-		network.PublicNetwork,
-		builder.BaseConfig.BindAddr,
-		networkKey,
-		builder.SporkID,
-		builder.IdentityProvider,
-		&builder.FlowConfig.NetworkConfig.ResourceManager,
-		p2pbuilderconfig.PeerManagerDisableConfig(), // disable peer manager for follower
-		&p2p.DisallowListCacheConfig{
+		NetworkingType:             network.PublicNetwork,
+		Address:                    builder.BaseConfig.BindAddr,
+		NetworkKey:                 networkKey,
+		SporkId:                    builder.SporkID,
+		IdProvider:                 builder.IdentityProvider,
+		ResourceManagerParams:      &builder.FlowConfig.NetworkConfig.ResourceManager,
+		RpcInspectorParams:         &builder.FlowConfig.NetworkConfig.GossipSub.RpcInspector,
+		PeerManagerParams:          p2pbuilderconfig.PeerManagerDisableConfig(),
+		SubscriptionProviderParams: &builder.FlowConfig.NetworkConfig.GossipSub.SubscriptionProvider,
+		DisallowListCacheCfg: &p2p.DisallowListCacheConfig{
 			MaxSize: builder.FlowConfig.NetworkConfig.DisallowListNotificationCacheSize,
 			Metrics: metrics.DisallowListCacheMetricsFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
 		},
-		&p2pbuilderconfig.UnicastConfig{
-			Unicast: builder.FlowConfig.NetworkConfig.Unicast,
-		}).
+		UnicastConfig: &p2pbuilderconfig.UnicastConfig{
+			Unicast:                builder.FlowConfig.NetworkConfig.Unicast,
+			RateLimiterDistributor: builder.UnicastRateLimiterDistributor,
+		},
+		GossipSubCfg: &builder.FlowConfig.NetworkConfig.GossipSub,
+	}
+	nodeBuilder, err := p2pbuilder.NewNodeBuilder(params)
+	if err != nil {
+		return nil, fmt.Errorf("could not create libp2p node builder: %w", err)
+	}
+	libp2pNode, err := nodeBuilder.
 		SetSubscriptionFilter(
 			subscription.NewRoleBasedFilter(
 				subscription.UnstakedRole, builder.IdentityProvider,
@@ -610,7 +619,7 @@ func (builder *FollowerServiceBuilder) initPublicLibp2pNode(networkKey crypto.Pr
 		return nil, fmt.Errorf("could not build public libp2p node: %w", err)
 	}
 
-	builder.LibP2PNode = node
+	builder.LibP2PNode = libp2pNode
 
 	return builder.LibP2PNode, nil
 }

insecure/corruptlibp2p/libp2p_node_factory.go

@@ -80,25 +80,28 @@ func InitCorruptLibp2pNode(
 	if err != nil {
 		return nil, fmt.Errorf("could not get dht system activation status: %w", err)
 	}
-	builder, err := p2pbuilder.DefaultNodeBuilder(
-		log,
-		address,
-		network.PrivateNetwork,
-		flowKey,
-		sporkId,
-		idProvider,
-		metCfg,
+	params := &p2pbuilder.LibP2PNodeBuilderConfig{
+		Logger:                     log,
+		MetricsConfig:              metCfg,
+		NetworkingType:             network.PrivateNetwork,
+		Address:                    address,
+		NetworkKey:                 flowKey,
+		SporkId:                    sporkId,
+		IdProvider:                 idProvider,
+		ResourceManagerParams:      &netConfig.ResourceManager,
+		RpcInspectorParams:         &netConfig.GossipSub.RpcInspector,
+		PeerManagerParams:          peerManagerCfg,
+		SubscriptionProviderParams: &netConfig.GossipSub.SubscriptionProvider,
+		DisallowListCacheCfg:       disallowListCacheCfg,
+		UnicastConfig:              uniCfg,
+		GossipSubCfg:               &netConfig.GossipSub,
+	}
+	builder, err := p2pbuilder.DefaultNodeBuilder(params,
 		resolver,
 		role,
 		connGaterCfg,
-		peerManagerCfg,
-		&netConfig.GossipSub,
-		&netConfig.ResourceManager,
-		uniCfg,
 		&netConfig.ConnectionManager,
-		disallowListCacheCfg,
 		dhtActivationStatus)
-
 	if err != nil {
 		return nil, fmt.Errorf("could not create corrupt libp2p node builder: %w", err)
 	}