[Networking] Report message decode failures on staked channels to ALSP

network/underlay/network.go

@@ -1260,19 +1260,43 @@ func (n *Network) processAuthenticatedMessage(msg *message.Message, peerID peer.
 	}
 
 	channel := channels.Channel(msg.ChannelID)
+
+	// identity lazily resolves the full identity of the authenticated peer for staked channels.
+	// The lookup is deferred so it only runs on violation paths; the slashing consumer uses a nil
+	// Identity to skip ALSP reporting on public channels, where identities are not treated as
+	// staked participants.
+	identity := func() *flow.Identity {
+		if channels.IsPublicChannel(channel) {
+			return nil
+		}
+		id, ok := n.Identity(peerID)
+		if !ok {
+			// On a staked channel the peer has already cleared sender authorization, so a
+			// missing identity here means it vanished between those checks and now. Log as
+			// suspicious rather than silently skipping the ALSP report.
+			n.logger.Warn().
+				Str("peer_id", p2plogging.PeerId(peerID)).
+				Str("channel", channel.String()).
+				Bool(logging.KeySuspicious, true).
+				Msg("could not resolve identity of authenticated peer on staked channel")
+			return nil
+		}
+		return id
+	}
+
 	decodedMsgPayload, err := n.codec.Decode(msg.Payload)
 	switch {
 	case codec.IsErrUnknownMsgCode(err):
 		// slash peer if message contains unknown message code byte
 		violation := &network.Violation{
-			PeerID: p2plogging.PeerId(peerID), OriginID: originId, Channel: channel, Protocol: protocol, Err: err,
+			Identity: identity(), PeerID: p2plogging.PeerId(peerID), OriginID: originId, Channel: channel, Protocol: protocol, Err: err,
 		}
 		n.slashingViolationsConsumer.OnUnknownMsgTypeError(violation)
 		return
 	case codec.IsErrMsgUnmarshal(err) || codec.IsErrInvalidEncoding(err):
 		// slash if peer sent a message that could not be marshalled into the message type denoted by the message code byte
 		violation := &network.Violation{
-			PeerID: p2plogging.PeerId(peerID), OriginID: originId, Channel: channel, Protocol: protocol, Err: err,
+			Identity: identity(), PeerID: p2plogging.PeerId(peerID), OriginID: originId, Channel: channel, Protocol: protocol, Err: err,
 		}
 		n.slashingViolationsConsumer.OnInvalidMsgError(violation)
 		return
@@ -1282,7 +1306,7 @@ func (n *Network) processAuthenticatedMessage(msg *message.Message, peerID peer.
 		// collect slashing data because this could potentially lead to slashing
 		err = fmt.Errorf("unexpected error during message validation: %w", err)
 		violation := &network.Violation{
-			PeerID: p2plogging.PeerId(peerID), OriginID: originId, Channel: channel, Protocol: protocol, Err: err,
+			Identity: identity(), PeerID: p2plogging.PeerId(peerID), OriginID: originId, Channel: channel, Protocol: protocol, Err: err,
 		}
 		n.slashingViolationsConsumer.OnUnexpectedError(violation)
 		return
@@ -1292,7 +1316,7 @@ func (n *Network) processAuthenticatedMessage(msg *message.Message, peerID peer.
 	if err != nil {
 		err = fmt.Errorf("failed to convert message to internal: %w", err)
 		violation := &network.Violation{
-			PeerID: p2plogging.PeerId(peerID), OriginID: originId, Channel: channel, Protocol: protocol, Err: err,
+			Identity: identity(), PeerID: p2plogging.PeerId(peerID), OriginID: originId, Channel: channel, Protocol: protocol, Err: err,
 		}
 		n.slashingViolationsConsumer.OnInvalidMsgError(violation)
 		return

network/underlay/network_test.go

@@ -1,18 +1,27 @@
 package underlay
 
 import (
+	"bytes"
 	"testing"
 
+	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/model/messages"
 	modulemock "github.com/onflow/flow-go/module/mock"
 	"github.com/onflow/flow-go/network"
+	"github.com/onflow/flow-go/network/alsp"
+	"github.com/onflow/flow-go/network/channels"
+	"github.com/onflow/flow-go/network/codec"
+	"github.com/onflow/flow-go/network/codec/cbor"
 	"github.com/onflow/flow-go/network/message"
 	mockmsg "github.com/onflow/flow-go/network/mock"
 	p2plogging "github.com/onflow/flow-go/network/p2p/logging"
+	"github.com/onflow/flow-go/network/slashing"
 	"github.com/onflow/flow-go/network/validator"
 	"github.com/onflow/flow-go/utils/unittest"
 )
@@ -145,3 +154,143 @@ func TestGetAuthorizedIdentity_ActivePeer(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, activeIdentity, identity)
 }
+
+// stubIDTranslator implements the p2p.IDTranslator interface used by the Network for tests.
+// It only needs to translate a peer ID back to the configured Flow ID.
+type stubIDTranslator struct {
+	id flow.Identifier
+}
+
+func (t stubIDTranslator) GetPeerID(_ flow.Identifier) (peer.ID, error) {
+	return "", nil
+}
+
+func (t stubIDTranslator) GetFlowID(_ peer.ID) (flow.Identifier, error) {
+	return t.id, nil
+}
+
+// newTestNetwork returns a Network with the minimal field set needed to drive
+// processAuthenticatedMessage in tests.
+func newTestNetwork(
+	t *testing.T,
+	idProvider *modulemock.IdentityProvider,
+	metrics *modulemock.NetworkSecurityMetrics,
+	reportConsumer *mockmsg.MisbehaviorReportConsumer,
+	originID flow.Identifier,
+) *Network {
+	return &Network{
+		logger:                     unittest.Logger(),
+		identityProvider:           idProvider,
+		identityTranslator:         stubIDTranslator{id: originID},
+		codec:                      cbor.NewCodec(),
+		slashingViolationsConsumer: slashing.NewSlashingViolationsConsumer(unittest.Logger(), metrics, reportConsumer),
+	}
+}
+
+// TestProcessAuthenticatedMessage_ReportsDecodeFailureOnStakedChannel verifies that a staked peer
+// sending a message with a valid message-code byte but an undecodable payload is reported to ALSP
+// so that a penalty can be applied.
+func TestProcessAuthenticatedMessage_ReportsDecodeFailureOnStakedChannel(t *testing.T) {
+	attackerIdentity := unittest.IdentityFixture(
+		unittest.WithRole(flow.RoleConsensus),
+		unittest.WithParticipationStatus(flow.EpochParticipationStatusActive),
+	)
+	attackerPeerID := unittest.PeerIdFixture(t)
+	channel := channels.ConsensusCommittee
+
+	// A valid message-code byte followed by undecodable CBOR garbage. This passes the
+	// authorized-sender check (which only looks at the code byte) but fails codec.Decode.
+	payload := append([]byte{codec.CodeBlockProposal.Uint8()}, bytes.Repeat([]byte{0xFF}, 32)...)
+
+	reportConsumer := mockmsg.NewMisbehaviorReportConsumer(t)
+	var reported []network.MisbehaviorReport
+	reportConsumer.On("ReportMisbehaviorOnChannel", channel, mock.Anything).
+		Run(func(args mock.Arguments) {
+			reported = append(reported, args.Get(1).(network.MisbehaviorReport))
+		}).
+		Once()
+
+	metrics := modulemock.NewNetworkSecurityMetrics(t)
+	metrics.On("OnUnauthorizedMessage", attackerIdentity.Role.String(), "unknown", channel.String(), alsp.InvalidMessage.String()).Once()
+
+	idProvider := modulemock.NewIdentityProvider(t)
+	idProvider.On("ByPeerID", attackerPeerID).Return(attackerIdentity, true).Once()
+
+	net := newTestNetwork(t, idProvider, metrics, reportConsumer, attackerIdentity.NodeID)
+
+	net.processAuthenticatedMessage(&message.Message{ChannelID: channel.String(), Payload: payload}, attackerPeerID, message.ProtocolTypePubSub)
+
+	require.Len(t, reported, 1, "ALSP report must be submitted for a staked peer that sends an undecodable message")
+	require.Equal(t, attackerIdentity.NodeID, reported[0].OriginId())
+	require.Equal(t, alsp.InvalidMessage, reported[0].Reason())
+}
+
+// TestProcessAuthenticatedMessage_ReportsToInternalFailureOnStakedChannel verifies that a staked
+// peer sending well-formed CBOR that decodes successfully but fails structural validation in
+// ToInternal (e.g. a proposal with an empty chain ID) is also reported to ALSP.
+func TestProcessAuthenticatedMessage_ReportsToInternalFailureOnStakedChannel(t *testing.T) {
+	attackerIdentity := unittest.IdentityFixture(
+		unittest.WithRole(flow.RoleConsensus),
+		unittest.WithParticipationStatus(flow.EpochParticipationStatusActive),
+	)
+	attackerPeerID := unittest.PeerIdFixture(t)
+	channel := channels.ConsensusCommittee
+
+	// Well-formed CBOR that decodes into a proposal but fails structural validation in
+	// ToInternal (the empty proposal has an empty chain ID). Codec.Encode already prepends
+	// the message-code byte, so no manual prepend here.
+	cborCodec := cbor.NewCodec()
+	payload, err := cborCodec.Encode(&messages.Proposal{})
+	require.NoError(t, err)
+
+	// Pin the payload's properties: it must decode cleanly and fail in ToInternal, otherwise
+	// this test silently covers the unmarshal branch instead.
+	decoded, err := cborCodec.Decode(payload)
+	require.NoError(t, err, "payload must decode; this test covers the ToInternal branch")
+	_, err = decoded.ToInternal()
+	require.Error(t, err, "payload must fail structural validation in ToInternal")
+
+	reportConsumer := mockmsg.NewMisbehaviorReportConsumer(t)
+	var reported []network.MisbehaviorReport
+	reportConsumer.On("ReportMisbehaviorOnChannel", channel, mock.Anything).
+		Run(func(args mock.Arguments) {
+			reported = append(reported, args.Get(1).(network.MisbehaviorReport))
+		}).
+		Once()
+
+	metrics := modulemock.NewNetworkSecurityMetrics(t)
+	metrics.On("OnUnauthorizedMessage", attackerIdentity.Role.String(), "unknown", channel.String(), alsp.InvalidMessage.String()).Once()
+
+	idProvider := modulemock.NewIdentityProvider(t)
+	idProvider.On("ByPeerID", attackerPeerID).Return(attackerIdentity, true).Once()
+
+	net := newTestNetwork(t, idProvider, metrics, reportConsumer, attackerIdentity.NodeID)
+
+	net.processAuthenticatedMessage(&message.Message{ChannelID: channel.String(), Payload: payload}, attackerPeerID, message.ProtocolTypePubSub)
+
+	require.Len(t, reported, 1, "ALSP report must be submitted for a staked peer that sends a structurally invalid message")
+	require.Equal(t, attackerIdentity.NodeID, reported[0].OriginId())
+	require.Equal(t, alsp.InvalidMessage, reported[0].Reason())
+}
+
+// TestProcessAuthenticatedMessage_SkipsPublicChannelDecodeFailure verifies that decode failures on
+// public channels are not reported to ALSP, preserving the existing exemption for the public network.
+func TestProcessAuthenticatedMessage_SkipsPublicChannelDecodeFailure(t *testing.T) {
+	peerID := unittest.PeerIdFixture(t)
+	channel := channels.PublicReceiveBlocks
+
+	// Same undecodable payload as the staked-channel test.
+	payload := append([]byte{codec.CodeBlockProposal.Uint8()}, bytes.Repeat([]byte{0xFF}, 32)...)
+
+	reportConsumer := mockmsg.NewMisbehaviorReportConsumer(t)
+	// No ReportMisbehaviorOnChannel expectation: mock strictness is the assertion that the
+	// violation must be skipped for public channels.
+
+	metrics := modulemock.NewNetworkSecurityMetrics(t)
+	metrics.On("OnUnauthorizedMessage", "unknown", "unknown", channel.String(), alsp.InvalidMessage.String()).Once()
+	metrics.On("OnViolationReportSkipped").Once()
+
+	net := newTestNetwork(t, modulemock.NewIdentityProvider(t), metrics, reportConsumer, unittest.IdentifierFixture())
+
+	net.processAuthenticatedMessage(&message.Message{ChannelID: channel.String(), Payload: payload}, peerID, message.ProtocolTypePubSub)
+}