Merge pull request #4512 from onflow/khalil/6743-slashing-violations-…

…consumer-alsp

Khalil/6743 Slashing Violations Consumer ALSP misbehavior report integration

Makefile

@@ -167,7 +167,7 @@ generate-mocks: install-mock-generators
 	rm -rf ./fvm/environment/mock
 	mockery --name '.*' --dir=fvm/environment --case=underscore --output="./fvm/environment/mock" --outpkg="mock"
 	mockery --name '.*' --dir=ledger --case=underscore --output="./ledger/mock" --outpkg="mock"
-	mockery --name 'ViolationsConsumer' --dir=network/slashing --case=underscore --output="./network/mocknetwork" --outpkg="mocknetwork"
+	mockery --name 'ViolationsConsumer' --dir=network --case=underscore --output="./network/mocknetwork" --outpkg="mocknetwork"
 	mockery --name '.*' --dir=network/p2p/ --case=underscore --output="./network/p2p/mock" --outpkg="mockp2p"
 	mockery --name '.*' --dir=network/alsp --case=underscore --output="./network/alsp/mock" --outpkg="mockalsp"
 	mockery --name 'Vertex' --dir="./module/forest" --case=underscore --output="./module/forest/mock" --outpkg="mock"

cmd/access/node_builder/access_node_builder.go

@@ -77,7 +77,6 @@ import (
 	"github.com/onflow/flow-go/network/p2p/translator"
 	"github.com/onflow/flow-go/network/p2p/unicast/protocols"
 	relaynet "github.com/onflow/flow-go/network/relay"
-	"github.com/onflow/flow-go/network/slashing"
 	"github.com/onflow/flow-go/network/topology"
 	"github.com/onflow/flow-go/network/validator"
 	"github.com/onflow/flow-go/state/protocol"
@@ -1261,15 +1260,14 @@ func (builder *FlowAccessNodeBuilder) initMiddleware(nodeID flow.Identifier,
 ) network.Middleware {
 	logger := builder.Logger.With().Bool("staked", false).Logger()
 	mw := middleware.NewMiddleware(&middleware.Config{
-		Logger:                     logger,
-		Libp2pNode:                 libp2pNode,
-		FlowId:                     nodeID,
-		BitSwapMetrics:             builder.Metrics.Bitswap,
-		RootBlockID:                builder.SporkID,
-		UnicastMessageTimeout:      middleware.DefaultUnicastTimeout,
-		IdTranslator:               builder.IDTranslator,
-		Codec:                      builder.CodecFactory(),
-		SlashingViolationsConsumer: slashing.NewSlashingViolationsConsumer(logger, networkMetrics),
+		Logger:                logger,
+		Libp2pNode:            libp2pNode,
+		FlowId:                nodeID,
+		BitSwapMetrics:        builder.Metrics.Bitswap,
+		RootBlockID:           builder.SporkID,
+		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
+		IdTranslator:          builder.IDTranslator,
+		Codec:                 builder.CodecFactory(),
 	},
 		middleware.WithMessageValidators(validators...), // use default identifier provider
 	)

cmd/observer/node_builder/observer_builder.go

@@ -62,7 +62,6 @@ import (
 	"github.com/onflow/flow-go/network/p2p/translator"
 	"github.com/onflow/flow-go/network/p2p/unicast/protocols"
 	"github.com/onflow/flow-go/network/p2p/utils"
-	"github.com/onflow/flow-go/network/slashing"
 	"github.com/onflow/flow-go/network/validator"
 	stateprotocol "github.com/onflow/flow-go/state/protocol"
 	badgerState "github.com/onflow/flow-go/state/protocol/badger"
@@ -905,17 +904,15 @@ func (builder *ObserverServiceBuilder) initMiddleware(nodeID flow.Identifier,
 	libp2pNode p2p.LibP2PNode,
 	validators ...network.MessageValidator,
 ) network.Middleware {
-	slashingViolationsConsumer := slashing.NewSlashingViolationsConsumer(builder.Logger, builder.Metrics.Network)
 	mw := middleware.NewMiddleware(&middleware.Config{
-		Logger:                     builder.Logger,
-		Libp2pNode:                 libp2pNode,
-		FlowId:                     nodeID,
-		BitSwapMetrics:             builder.Metrics.Bitswap,
-		RootBlockID:                builder.SporkID,
-		UnicastMessageTimeout:      middleware.DefaultUnicastTimeout,
-		IdTranslator:               builder.IDTranslator,
-		Codec:                      builder.CodecFactory(),
-		SlashingViolationsConsumer: slashingViolationsConsumer,
+		Logger:                builder.Logger,
+		Libp2pNode:            libp2pNode,
+		FlowId:                nodeID,
+		BitSwapMetrics:        builder.Metrics.Bitswap,
+		RootBlockID:           builder.SporkID,
+		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
+		IdTranslator:          builder.IDTranslator,
+		Codec:                 builder.CodecFactory(),
 	},
 		middleware.WithMessageValidators(validators...), // use default identifier provider
 	)

cmd/scaffold.go

@@ -60,7 +60,6 @@ import (
 	"github.com/onflow/flow-go/network/p2p/unicast/protocols"
 	"github.com/onflow/flow-go/network/p2p/unicast/ratelimit"
 	"github.com/onflow/flow-go/network/p2p/utils/ratelimiter"
-	"github.com/onflow/flow-go/network/slashing"
 	"github.com/onflow/flow-go/network/topology"
 	"github.com/onflow/flow-go/state/protocol"
 	badgerState "github.com/onflow/flow-go/state/protocol/badger"
@@ -437,17 +436,15 @@ func (fnb *FlowNodeBuilder) InitFlowNetworkWithConduitFactory(
 		mwOpts = append(mwOpts, middleware.WithPeerManagerFilters(peerManagerFilters))
 	}
 
-	slashingViolationsConsumer := slashing.NewSlashingViolationsConsumer(fnb.Logger, fnb.Metrics.Network)
 	mw := middleware.NewMiddleware(&middleware.Config{
-		Logger:                     fnb.Logger,
-		Libp2pNode:                 fnb.LibP2PNode,
-		FlowId:                     fnb.Me.NodeID(),
-		BitSwapMetrics:             fnb.Metrics.Bitswap,
-		RootBlockID:                fnb.SporkID,
-		UnicastMessageTimeout:      fnb.BaseConfig.FlowConfig.NetworkConfig.UnicastMessageTimeout,
-		IdTranslator:               fnb.IDTranslator,
-		Codec:                      fnb.CodecFactory(),
-		SlashingViolationsConsumer: slashingViolationsConsumer,
+		Logger:                fnb.Logger,
+		Libp2pNode:            fnb.LibP2PNode,
+		FlowId:                fnb.Me.NodeID(),
+		BitSwapMetrics:        fnb.Metrics.Bitswap,
+		RootBlockID:           fnb.SporkID,
+		UnicastMessageTimeout: fnb.FlowConfig.NetworkConfig.UnicastMessageTimeout,
+		IdTranslator:          fnb.IDTranslator,
+		Codec:                 fnb.CodecFactory(),
 	},
 		mwOpts...)
 

follower/follower_builder.go

@@ -56,7 +56,6 @@ import (
 	"github.com/onflow/flow-go/network/p2p/translator"
 	"github.com/onflow/flow-go/network/p2p/unicast/protocols"
 	"github.com/onflow/flow-go/network/p2p/utils"
-	"github.com/onflow/flow-go/network/slashing"
 	"github.com/onflow/flow-go/network/validator"
 	"github.com/onflow/flow-go/state/protocol"
 	badgerState "github.com/onflow/flow-go/state/protocol/badger"
@@ -748,15 +747,14 @@ func (builder *FollowerServiceBuilder) initMiddleware(nodeID flow.Identifier,
 	validators ...network.MessageValidator,
 ) network.Middleware {
 	mw := middleware.NewMiddleware(&middleware.Config{
-		Logger:                     builder.Logger,
-		Libp2pNode:                 libp2pNode,
-		FlowId:                     nodeID,
-		BitSwapMetrics:             builder.Metrics.Bitswap,
-		RootBlockID:                builder.SporkID,
-		UnicastMessageTimeout:      middleware.DefaultUnicastTimeout,
-		IdTranslator:               builder.IDTranslator,
-		Codec:                      builder.CodecFactory(),
-		SlashingViolationsConsumer: slashing.NewSlashingViolationsConsumer(builder.Logger, builder.Metrics.Network),
+		Logger:                builder.Logger,
+		Libp2pNode:            libp2pNode,
+		FlowId:                nodeID,
+		BitSwapMetrics:        builder.Metrics.Bitswap,
+		RootBlockID:           builder.SporkID,
+		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
+		IdTranslator:          builder.IDTranslator,
+		Codec:                 builder.CodecFactory(),
 	},
 		middleware.WithMessageValidators(validators...),
 	)

module/metrics.go

@@ -42,6 +42,10 @@ type NetworkSecurityMetrics interface {
 
 	// OnRateLimitedPeer tracks the number of rate limited unicast messages seen on the network.
 	OnRateLimitedPeer(pid peer.ID, role, msgType, topic, reason string)
+
+	// OnViolationReportSkipped tracks the number of slashing violations consumer violations that were not
+	// reported for misbehavior when the identity of the sender not known.
+	OnViolationReportSkipped()
 }
 
 // GossipSubRouterMetrics encapsulates the metrics collectors for GossipSubRouter module of the networking layer.
@@ -182,6 +186,8 @@ type NetworkInboundQueueMetrics interface {
 type NetworkCoreMetrics interface {
 	NetworkInboundQueueMetrics
 	AlspMetrics
+	NetworkSecurityMetrics
+
 	// OutboundMessageSent collects metrics related to a message sent by the node.
 	OutboundMessageSent(sizeBytes int, topic string, protocol string, messageType string)
 	// InboundMessageReceived collects metrics related to a message received by the node.
@@ -223,7 +229,6 @@ type AlspMetrics interface {
 // NetworkMetrics is the blanket abstraction that encapsulates the metrics collectors for the networking layer.
 type NetworkMetrics interface {
 	LibP2PMetrics
-	NetworkSecurityMetrics
 	NetworkCoreMetrics
 }
 

module/metrics/namespaces.go

@@ -28,6 +28,7 @@ const (
 	subsystemAuth         = "authorization"
 	subsystemRateLimiting = "ratelimit"
 	subsystemAlsp         = "alsp"
+	subsystemSecurity     = "security"
 )
 
 // Storage subsystems represent the various components of the storage layer.

module/metrics/network.go

@@ -45,9 +45,10 @@ type NetworkCollector struct {
 	dnsLookupRequestDroppedCount prometheus.Counter
 	routingTableSize             prometheus.Gauge
 
-	// authorization, rate limiting metrics
+	// security metrics
 	unAuthorizedMessagesCount       *prometheus.CounterVec
 	rateLimitedUnicastMessagesCount *prometheus.CounterVec
+	violationReportSkippedCount     prometheus.Counter
 
 	prefix string
 }
@@ -245,6 +246,15 @@ func NewNetworkCollector(logger zerolog.Logger, opts ...NetworkCollectorOpt) *Ne
 		}, []string{LabelNodeRole, LabelMessage, LabelChannel, LabelRateLimitReason},
 	)
 
+	nc.violationReportSkippedCount = promauto.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: namespaceNetwork,
+			Subsystem: subsystemSecurity,
+			Name:      nc.prefix + "slashing_violation_reports_skipped_count",
+			Help:      "number of slashing violations consumer violations that were not reported for misbehavior because the identity of the sender not known",
+		},
+	)
+
 	return nc
 }
 
@@ -358,3 +368,9 @@ func (nc *NetworkCollector) OnRateLimitedPeer(peerID peer.ID, role, msgType, top
 		Msg("unicast peer rate limited")
 	nc.rateLimitedUnicastMessagesCount.WithLabelValues(role, msgType, topic, reason).Inc()
 }
+
+// OnViolationReportSkipped tracks the number of slashing violations consumer violations that were not
+// reported for misbehavior when the identity of the sender not known.
+func (nc *NetworkCollector) OnViolationReportSkipped() {
+	nc.violationReportSkippedCount.Inc()
+}

module/metrics/noop.go

@@ -303,3 +303,4 @@ func (nc *NoopCollector) AsyncProcessingStarted(string)
 func (nc *NoopCollector) AsyncProcessingFinished(string, time.Duration)             {}
 
 func (nc *NoopCollector) OnMisbehaviorReported(string, string) {}
+func (nc *NoopCollector) OnViolationReportSkipped()            {}

network/alsp/manager/manager_test.go

@@ -54,7 +54,7 @@ func TestNetworkPassesReportedMisbehavior(t *testing.T) {
 	misbehaviorReportManger.On("Ready").Return(readyDoneChan).Once()
 	misbehaviorReportManger.On("Done").Return(readyDoneChan).Once()
 	ids, nodes, _ := testutils.LibP2PNodeForMiddlewareFixture(t, 1)
-	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t))
+	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t), mocknetwork.NewViolationsConsumer(t))
 
 	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, mws[0])
 	net, err := p2p.NewNetwork(networkCfg, p2p.WithAlspManager(misbehaviorReportManger))
@@ -111,7 +111,7 @@ func TestHandleReportedMisbehavior_Cache_Integration(t *testing.T) {
 		}),
 	}
 	ids, nodes, _ := testutils.LibP2PNodeForMiddlewareFixture(t, 1)
-	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t))
+	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t), mocknetwork.NewViolationsConsumer(t))
 	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, mws[0], p2p.WithAlspConfig(cfg))
 	net, err := p2p.NewNetwork(networkCfg)
 	require.NoError(t, err)
@@ -206,7 +206,7 @@ func TestHandleReportedMisbehavior_And_DisallowListing_Integration(t *testing.T)
 
 	ids, nodes, _ := testutils.LibP2PNodeForMiddlewareFixture(t, 3,
 		p2ptest.WithPeerManagerEnabled(p2ptest.PeerManagerConfigFixture(), nil))
-	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t))
+	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t), mocknetwork.NewViolationsConsumer(t))
 	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, mws[0], p2p.WithAlspConfig(cfg))
 	victimNetwork, err := p2p.NewNetwork(networkCfg)
 	require.NoError(t, err)
@@ -279,7 +279,7 @@ func TestMisbehaviorReportMetrics(t *testing.T) {
 	cfg.AlspMetrics = alspMetrics
 
 	ids, nodes, _ := testutils.LibP2PNodeForMiddlewareFixture(t, 1)
-	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t))
+	mws, _ := testutils.MiddlewareFixtures(t, ids, nodes, testutils.MiddlewareConfigFixture(t), mocknetwork.NewViolationsConsumer(t))
 	networkCfg := testutils.NetworkConfigFixture(t, *ids[0], ids, mws[0], p2p.WithAlspConfig(cfg))
 	net, err := p2p.NewNetwork(networkCfg)
 	require.NoError(t, err)

network/alsp/misbehavior.go

@@ -24,6 +24,25 @@ const (
 	// the message is not valid according to the engine's validation logic. The decision to consider a message invalid
 	// is up to the engine.
 	InvalidMessage network.Misbehavior = "misbehavior-invalid-message"
+
+	// UnExpectedValidationError is a misbehavior that is reported when a validation error is encountered during message validation before the message
+	// is processed by an engine.
+	UnExpectedValidationError network.Misbehavior = "unexpected-validation-error"
+
+	// UnknownMsgType is a misbehavior that is reported when a message of unknown type is received from a peer.
+	UnknownMsgType network.Misbehavior = "unknown-message-type"
+
+	// SenderEjected is a misbehavior that is reported when a message is received from an ejected peer.
+	SenderEjected network.Misbehavior = "sender-ejected"
+
+	// UnauthorizedUnicastOnChannel is a misbehavior that is reported when a message not authorized to be sent via unicast is received via unicast.
+	UnauthorizedUnicastOnChannel network.Misbehavior = "unauthorized-unicast-on-channel"
+
+	// UnAuthorizedSender is a misbehavior that is reported when a message is sent by an unauthorized role.
+	UnAuthorizedSender network.Misbehavior = "unauthorized-sender"
+
+	// UnauthorizedPublishOnChannel is a misbehavior that is reported when a message not authorized to be sent via pubsub is received via pubsub.
+	UnauthorizedPublishOnChannel network.Misbehavior = "unauthorized-pubsub-on-channel"
 )
 
 func AllMisbehaviorTypes() []network.Misbehavior {
@@ -33,5 +52,11 @@ func AllMisbehaviorTypes() []network.Misbehavior {
 		RedundantMessage,
 		UnsolicitedMessage,
 		InvalidMessage,
+		UnExpectedValidationError,
+		UnknownMsgType,
+		SenderEjected,
+		UnauthorizedUnicastOnChannel,
+		UnauthorizedPublishOnChannel,
+		UnAuthorizedSender,
 	}
 }

network/internal/testutils/testUtil.go

@@ -28,7 +28,6 @@ import (
 	netcache "github.com/onflow/flow-go/network/cache"
 	"github.com/onflow/flow-go/network/channels"
 	"github.com/onflow/flow-go/network/codec/cbor"
-	"github.com/onflow/flow-go/network/mocknetwork"
 	"github.com/onflow/flow-go/network/netconf"
 	"github.com/onflow/flow-go/network/p2p"
 	"github.com/onflow/flow-go/network/p2p/conduit"
@@ -167,12 +166,11 @@ func LibP2PNodeForMiddlewareFixture(t *testing.T, n int, opts ...p2ptest.NodeFix
 // - a middleware config.
 func MiddlewareConfigFixture(t *testing.T) *middleware.Config {
 	return &middleware.Config{
-		Logger:                     unittest.Logger(),
-		BitSwapMetrics:             metrics.NewNoopCollector(),
-		RootBlockID:                sporkID,
-		UnicastMessageTimeout:      middleware.DefaultUnicastTimeout,
-		Codec:                      unittest.NetworkCodec(),
-		SlashingViolationsConsumer: mocknetwork.NewViolationsConsumer(t),
+		Logger:                unittest.Logger(),
+		BitSwapMetrics:        metrics.NewNoopCollector(),
+		RootBlockID:           sporkID,
+		UnicastMessageTimeout: middleware.DefaultUnicastTimeout,
+		Codec:                 unittest.NetworkCodec(),
 	}
 }
 
@@ -187,7 +185,7 @@ func MiddlewareConfigFixture(t *testing.T) *middleware.Config {
 // Returns:
 // - a list of middlewares - one for each identity.
 // - a list of UpdatableIDProvider - one for each identity.
-func MiddlewareFixtures(t *testing.T, identities flow.IdentityList, libP2PNodes []p2p.LibP2PNode, cfg *middleware.Config, opts ...middleware.OptionFn) ([]network.Middleware, []*unittest.UpdatableIDProvider) {
+func MiddlewareFixtures(t *testing.T, identities flow.IdentityList, libP2PNodes []p2p.LibP2PNode, cfg *middleware.Config, consumer network.ViolationsConsumer, opts ...middleware.OptionFn) ([]network.Middleware, []*unittest.UpdatableIDProvider) {
 	require.Equal(t, len(identities), len(libP2PNodes))
 
 	mws := make([]network.Middleware, len(identities))
@@ -199,8 +197,8 @@ func MiddlewareFixtures(t *testing.T, identities flow.IdentityList, libP2PNodes
 		cfg.FlowId = identities[i].NodeID
 		idProviders[i] = unittest.NewUpdatableIDProvider(identities)
 		cfg.IdTranslator = translator.NewIdentityProviderIDTranslator(idProviders[i])
-
 		mws[i] = middleware.NewMiddleware(cfg, opts...)
+		mws[i].SetSlashingViolationsConsumer(consumer)
 	}
 	return mws, idProviders
 }