Do not allow duplicate RemoveChannelIDs MERC-6645

llo/json_report_codec.go

@@ -81,6 +81,7 @@ func (cdc JSONReportCodec) Encode(ctx context.Context, r Report, _ llotypes.Chan
 	return json.Marshal(e)
 }
 
+// Fuzz testing: MERC-6522
 func (cdc JSONReportCodec) Decode(b []byte) (r Report, err error) {
 	type decode struct {
 		ConfigDigest                string

llo/onchain_config_codec.go

@@ -35,7 +35,7 @@ var _ OnchainConfigCodec = EVMOnchainConfigCodec{}
 // returned by EncodeValueInt192.
 type EVMOnchainConfigCodec struct{}
 
-// TODO: Needs fuzz testing - MERC-3524
+// TODO: Needs fuzz testing - MERC-6522
 func (EVMOnchainConfigCodec) Decode(b []byte) (OnchainConfig, error) {
 	if len(b) != onchainConfigEncodedLength {
 		return OnchainConfig{}, fmt.Errorf("unexpected length of OnchainConfig, expected %v, got %v", onchainConfigEncodedLength, len(b))
@@ -60,7 +60,7 @@ func (EVMOnchainConfigCodec) Decode(b []byte) (OnchainConfig, error) {
 	return o, nil
 }
 
-// TODO: Needs fuzz testing - MERC-3524
+// TODO: Needs fuzz testing - MERC-6522
 func (EVMOnchainConfigCodec) Encode(c OnchainConfig) ([]byte, error) {
 	if c.Version != onchainConfigVersion {
 		return nil, fmt.Errorf("unexpected version of OnchainConfig, expected %v, got %v", onchainConfigVersion, c.Version)

llo/plugin_codecs.go

@@ -41,7 +41,7 @@ func (c protoObservationCodec) Encode(obs Observation) (types.Observation, error
 				// Ignore nil values
 				continue
 			} else if err != nil {
-				return nil, err
+				return nil, fmt.Errorf("failed to encode observation: %w", err)
 			}
 			streamValues[id] = &LLOStreamValue{
 				Type:  sv.Type(),
@@ -84,7 +84,7 @@ func channelDefinitionsToProtoObservation(in llotypes.ChannelDefinitions) (out m
 }
 
 // TODO: Guard against untrusted inputs!
-// MERC-3524
+// MERC-6522
 func (c protoObservationCodec) Decode(b types.Observation) (Observation, error) {
 	pbuf := &LLOObservationProto{}
 	err := proto.Unmarshal(b, pbuf)
@@ -95,6 +95,11 @@ func (c protoObservationCodec) Decode(b types.Observation) (Observation, error)
 	if len(pbuf.RemoveChannelIDs) > 0 {
 		removeChannelIDs = make(map[llotypes.ChannelID]struct{}, len(pbuf.RemoveChannelIDs))
 		for _, id := range pbuf.RemoveChannelIDs {
+			if _, exists := removeChannelIDs[id]; exists {
+				// Byzantine behavior makes this observation invalid; a
+				// well-behaved node should never encode duplicates here
+				return Observation{}, fmt.Errorf("failed to decode observation; duplicate channel ID in RemoveChannelIDs: %d", id)
+			}
 			removeChannelIDs[id] = struct{}{}
 		}
 	}
@@ -124,7 +129,7 @@ func (c protoObservationCodec) Decode(b types.Observation) (Observation, error)
 }
 
 // TODO: Needs fuzz testing
-// MERC-3524
+// MERC-6522
 func channelDefinitionsFromProtoObservation(channelDefinitions map[uint32]*LLOChannelDefinitionProto) llotypes.ChannelDefinitions {
 	if len(channelDefinitions) == 0 {
 		return nil
@@ -209,7 +214,7 @@ func channelDefinitionsToProtoOutcome(in llotypes.ChannelDefinitions) (out []*LL
 }
 
 // TODO: Needs thorough unit testing of all paths including nil handling
-// MERC-3524
+// MERC-6522
 func StreamAggregatesToProtoOutcome(in StreamAggregates) (out []*LLOStreamAggregate, err error) {
 	if len(in) > 0 {
 		out = make([]*LLOStreamAggregate, 0, len(in))
@@ -260,7 +265,7 @@ func validAfterSecondsToProtoOutcome(in map[llotypes.ChannelID]uint32) (out []*L
 }
 
 // TODO: Guard against untrusted inputs!
-// MERC-3524
+// MERC-6522
 func (protoOutcomeCodec) Decode(b ocr3types.Outcome) (outcome Outcome, err error) {
 	pbuf := &LLOOutcomeProto{}
 	err = proto.Unmarshal(b, pbuf)
@@ -284,7 +289,7 @@ func (protoOutcomeCodec) Decode(b ocr3types.Outcome) (outcome Outcome, err error
 }
 
 // TODO: Needs fuzz testing
-// MERC-3524
+// MERC-6522
 func channelDefinitionsFromProtoOutcome(in []*LLOChannelIDAndDefinitionProto) (out llotypes.ChannelDefinitions) {
 	if len(in) > 0 {
 		out = make(map[llotypes.ChannelID]llotypes.ChannelDefinition, len(in))
@@ -307,7 +312,7 @@ func channelDefinitionsFromProtoOutcome(in []*LLOChannelIDAndDefinitionProto) (o
 }
 
 // TODO: Needs fuzz testing
-// MERC-3524
+// MERC-6522
 func streamAggregatesFromProtoOutcome(in []*LLOStreamAggregate) (out StreamAggregates, err error) {
 	if len(in) > 0 {
 		out = make(StreamAggregates, len(in))
@@ -329,7 +334,7 @@ func streamAggregatesFromProtoOutcome(in []*LLOStreamAggregate) (out StreamAggre
 }
 
 // TODO: Needs fuzz testing
-// MERC-3524
+// MERC-6522
 func validAfterSecondsFromProtoOutcome(in []*LLOChannelIDAndValidAfterSecondsProto) (out map[llotypes.ChannelID]uint32) {
 	if len(in) > 0 {
 		out = make(map[llotypes.ChannelID]uint32, len(in))

llo/plugin_codecs_test.go

@@ -6,6 +6,7 @@ import (
 	"github.com/shopspring/decimal"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
 
 	llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo"
 )
@@ -57,6 +58,25 @@ func Test_protoObservationCodec(t *testing.T) {
 
 		assert.Equal(t, expectedObs, obs2)
 	})
+	t.Run("decoding with invalid data", func(t *testing.T) {
+		t.Run("not a protobuf", func(t *testing.T) {
+			_, err := (protoObservationCodec{}).Decode([]byte("not a protobuf"))
+			require.Error(t, err)
+
+			assert.Contains(t, err.Error(), "cannot parse invalid wire-format data")
+		})
+		t.Run("duplicate RemoveChannelIDs", func(t *testing.T) {
+			pbuf := &LLOObservationProto{
+				RemoveChannelIDs: []uint32{1, 1},
+			}
+
+			obsBytes, err := proto.Marshal(pbuf)
+			require.NoError(t, err)
+
+			_, err = (protoObservationCodec{}).Decode(obsBytes)
+			require.EqualError(t, err, "failed to decode observation; duplicate channel ID in RemoveChannelIDs: 1")
+		})
+	})
 }
 
 func Test_protoOutcomeCodec(t *testing.T) {