Validate Max Aggregatable Exponent Consensus

consensus/action/capability.go

@@ -252,7 +252,11 @@ func (c *consensusCapability) Simple(ctx context.Context, metadata capabilities.
 
 	if err := decodeObservationType(lggr, input); err != nil {
 		responseAndMetadata := capabilities.ResponseAndMetadata[*valuespb.Value]{}
-		return &responseAndMetadata, caperrors.NewPublicSystemError(fmt.Errorf("failed to decode observation: %s", err), caperrors.InvalidArgument)
+		return &responseAndMetadata, caperrors.NewPublicSystemError(fmt.Errorf("failed to decode observation: %w", err), caperrors.InvalidArgument)
+	}
+	if err := validateSimpleConsensusInput(input); err != nil {
+		responseAndMetadata := capabilities.ResponseAndMetadata[*valuespb.Value]{}
+		return &responseAndMetadata, caperrors.NewPublicSystemError(fmt.Errorf("invalid consensus input: %w", err), caperrors.InvalidArgument)
 	}
 
 	lggr.Debugw("received simple consensus request", "metadata", metadata)
@@ -652,3 +656,16 @@ func decodeObservationType(lggr logger.Logger, input *sdk.SimpleConsensusInputs)
 	}
 	return nil
 }
+
+func validateSimpleConsensusInput(input *sdk.SimpleConsensusInputs) error {
+	if valueInput, ok := input.GetObservation().(*sdk.SimpleConsensusInputs_Value); ok {
+		if err := oracle.ValidateAggregatableValue(valueInput.Value); err != nil {
+			return fmt.Errorf("invalid observation value: %w", err)
+		}
+	}
+	if err := oracle.ValidateAggregatableValue(input.Default); err != nil {
+		return fmt.Errorf("invalid default value: %w", err)
+	}
+
+	return nil
+}

consensus/oracle/consensus_execution.go

@@ -27,10 +27,54 @@ var (
 	ErrDecimalExponentOutOfRange                         = errors.New("decimal exponent out of range")
 )
 
-// Comparing decimals rescales them to a common exponent, which costs time and memory
-// proportional to the exponent gap, so a single observation with an extreme exponent
-// could stall every node computing the median.
-const maxDecimalExponent = 1000
+// MaxAggregatableExponent bounds the base-10 exponent of an aggregatable value.
+// Comparing decimals with different exponents rescales them to a common exponent,
+// and that cost grows with the exponent gap. Every in-tree producer reports
+// exponent 0; the symmetric window leaves room for scaled values.
+const MaxAggregatableExponent = 128
+
+// ValidateAggregatableValue validates values used in aggregation and bounds
+// decimal exponents that could make comparisons allocate unbounded memory.
+func ValidateAggregatableValue(value *valuespb.Value) error {
+	if value == nil {
+		return nil
+	}
+
+	switch v := value.Value.(type) {
+	case nil:
+		// nil observations are valid; consensus falls back to the default
+		return nil
+	case *valuespb.Value_DecimalValue:
+		if exponent := v.DecimalValue.GetExponent(); exponent > MaxAggregatableExponent || exponent < -MaxAggregatableExponent {
+			return fmt.Errorf("%w: got %d, allowed [%d, %d]", ErrDecimalExponentOutOfRange, exponent, -MaxAggregatableExponent, MaxAggregatableExponent)
+		}
+	case *valuespb.Value_MapValue:
+		for _, field := range v.MapValue.GetFields() {
+			if err := ValidateAggregatableValue(field); err != nil {
+				return err
+			}
+		}
+	case *valuespb.Value_ListValue:
+		for _, field := range v.ListValue.GetFields() {
+			if err := ValidateAggregatableValue(field); err != nil {
+				return err
+			}
+		}
+	case *valuespb.Value_StringValue,
+		*valuespb.Value_BoolValue,
+		*valuespb.Value_BytesValue,
+		*valuespb.Value_Int64Value,
+		*valuespb.Value_BigintValue,
+		*valuespb.Value_TimeValue,
+		*valuespb.Value_Float64Value,
+		*valuespb.Value_Uint64Value:
+		return nil
+	default:
+		return fmt.Errorf("unsupported value type %T", value.Value)
+	}
+
+	return nil
+}
 
 // Constants for type names used in aggregation logic.
 var (
@@ -258,8 +302,8 @@ func handleMedianAggregation(
 			filtered,
 			func(val *valuespb.Value) (decimal.Decimal, error) {
 				var d decimal.Decimal
-				if exp := val.GetDecimalValue().GetExponent(); exp > maxDecimalExponent || exp < -maxDecimalExponent {
-					return d, fmt.Errorf("%w: %d", ErrDecimalExponentOutOfRange, exp)
+				if err := ValidateAggregatableValue(val); err != nil {
+					return d, err
 				}
 				v, err := values.FromProto(val)
 				if err != nil {

consensus/oracle/plugin/plugin_outcome.go

@@ -67,6 +67,13 @@ func (r *reportingPlugin) ValidateObservation(ctx context.Context, outctx ocr3ty
 			lggr.Warnw("observation missing input", "requestID", requestID)
 			return fmt.Errorf("observation from observer %d is missing input for request %s", ao.Observer, requestID)
 		}
+
+		if valueInput, ok := reqObs.Input.GetObservation().(*sdk.SimpleConsensusInputs_Value); ok {
+			if err := oracle.ValidateAggregatableValue(valueInput.Value); err != nil {
+				lggr.Warnw("observation contains an invalid aggregatable value", "requestID", requestID, "error", err)
+				return fmt.Errorf("observation from observer %d has invalid value for request %s: %w", ao.Observer, requestID, err)
+			}
+		}
 	}
 
 	return nil

consensus/oracle/plugin/plugin_outcome_test.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"testing"
 	"time"
 
+	"github.com/shopspring/decimal"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
@@ -24,6 +26,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-protos/cre/go/sdk"
 	"github.com/smartcontractkit/chainlink-protos/cre/go/values"
+	valuespb "github.com/smartcontractkit/chainlink-protos/cre/go/values/pb"
 
 	"github.com/smartcontractkit/libocr/commontypes"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
@@ -227,6 +230,52 @@ func Test_ValidateObservation_RequestIDConsistency(t *testing.T) {
 	require.ErrorContains(t, err, "not in the query")
 }
 
+func Test_ValidateObservation_RejectsExtremeDecimalExponent(t *testing.T) {
+	t.Parallel()
+
+	lggr := logger.Test(t)
+	reportingPlugin, _ := createReportingPlugin(t, lggr, 2, 7, 5, defaultMaxLengthBytes)
+
+	md := testMetaData()
+	reqID := md.RequestID()
+	qBytes, err := proto.Marshal(&oracletypes.Query{RequestIDs: []string{reqID}})
+	require.NoError(t, err)
+
+	ao := makeOutcomeTestObs(t, reqID, md, sdk.AggregationType_AGGREGATION_TYPE_MEDIAN, 0, false)
+	obs := &oracletypes.Observation{}
+	require.NoError(t, proto.Unmarshal(ao.Observation, obs))
+	obs.Observations[reqID].Input.Observation = &sdk.SimpleConsensusInputs_Value{
+		Value: valuespb.NewDecimalValue(decimal.New(1, math.MaxInt32)),
+	}
+	ao.Observation, err = proto.Marshal(obs)
+	require.NoError(t, err)
+
+	err = reportingPlugin.ValidateObservation(t.Context(), ocr3types.OutcomeContext{SeqNr: 1}, qBytes, ao)
+	require.Error(t, err)
+	require.ErrorContains(t, err, "decimal exponent out of range")
+}
+
+func Test_ValidateObservation_AcceptsNilValue(t *testing.T) {
+	t.Parallel()
+
+	lggr := logger.Test(t)
+	reportingPlugin, _ := createReportingPlugin(t, lggr, 2, 7, 5, defaultMaxLengthBytes)
+
+	md := testMetaData()
+	reqID := md.RequestID()
+	qBytes, err := proto.Marshal(&oracletypes.Query{RequestIDs: []string{reqID}})
+	require.NoError(t, err)
+
+	ao := makeOutcomeTestObs(t, reqID, md, sdk.AggregationType_AGGREGATION_TYPE_MEDIAN, 0, false)
+	obs := &oracletypes.Observation{}
+	require.NoError(t, proto.Unmarshal(ao.Observation, obs))
+	obs.Observations[reqID].Input.Observation = &sdk.SimpleConsensusInputs_Value{Value: &valuespb.Value{}}
+	ao.Observation, err = proto.Marshal(obs)
+	require.NoError(t, err)
+
+	require.NoError(t, reportingPlugin.ValidateObservation(t.Context(), ocr3types.OutcomeContext{SeqNr: 1}, qBytes, ao))
+}
+
 // Test_Outcome_PlusOneErrors checks that when f+1 errors are received, Outcome() embeds the
 // per-field metadata string ("Consensus metadata: requestId=...") and the descriptor type
 // string ("Descriptor type: AGGREGATION_TYPE_MEDIAN").