cre-5770: extra observations validation for reporting

pkg/capabilities/consensus/ocr3/identity_forge_test.go

@@ -0,0 +1,162 @@
+package ocr3
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/smartcontractkit/libocr/commontypes"
+	"github.com/smartcontractkit/libocr/offchainreporting2/types"
+	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
+
+	pbtypes "github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/ocr3/types"
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/requests"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-protos/cre/go/values"
+)
+
+type forgeAggregator struct{}
+
+func (forgeAggregator) Aggregate(_ logger.Logger, _ *pbtypes.AggregationOutcome, obs map[commontypes.OracleID][]values.Value, _ int) (*pbtypes.AggregationOutcome, error) {
+	var price string
+	for _, vals := range obs {
+		if len(vals) > 0 {
+			s, _ := vals[0].Unwrap()
+			price, _ = s.(string)
+			break
+		}
+	}
+	nm, err := values.NewMap(map[string]any{"price": price})
+	if err != nil {
+		return nil, err
+	}
+	return &pbtypes.AggregationOutcome{
+		EncodableOutcome: values.Proto(nm).GetMapValue(),
+		ShouldReport:     true,
+	}, nil
+}
+
+type forgeCapability struct {
+	aggWfID string
+	encWfID string
+}
+
+func (c *forgeCapability) GetAggregator(workflowID string) (pbtypes.Aggregator, error) {
+	c.aggWfID = workflowID
+	return forgeAggregator{}, nil
+}
+
+func (c *forgeCapability) GetEncoderByWorkflowID(workflowID string) (pbtypes.Encoder, error) {
+	c.encWfID = workflowID
+	return &enc{}, nil
+}
+
+func (c *forgeCapability) GetEncoderByName(string, *values.Map) (pbtypes.Encoder, error) {
+	return &enc{}, nil
+}
+
+func (c *forgeCapability) GetRegisteredWorkflowsIDs() []string { return nil }
+func (c *forgeCapability) UnregisterWorkflowID(string)         {}
+
+func TestReportingPlugin_LeaderQueryIdentityForgery(t *testing.T) {
+	const (
+		sharedExecID = "exec-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+
+		attackerWfID    = "attacker-workflow-id"
+		attackerWfOwner = "attacker-owner"
+		attackerWfName  = "attacker-name"
+		attackerReport  = "0002"
+		attackerKey     = "attacker-key"
+
+		victimWfID    = "victim-workflow-id"
+		victimWfOwner = "victim-owner"
+		victimWfName  = "victim-name"
+		victimReport  = "0001"
+		victimKey     = "victim-key"
+	)
+
+	obsVals, err := values.NewList([]any{"attacker-price"})
+	require.NoError(t, err)
+
+	mkObs := func(id *pbtypes.Id) []types.AttributedObservation {
+		obs := &pbtypes.Observations{
+			Observations: []*pbtypes.Observation{{
+				Id:           id,
+				Observations: values.Proto(obsVals).GetListValue(),
+			}},
+			RegisteredWorkflowIds: []string{attackerWfID, victimWfID},
+		}
+		raw, err := proto.Marshal(obs)
+		require.NoError(t, err)
+		return []types.AttributedObservation{
+			{Observation: raw, Observer: commontypes.OracleID(0)},
+			{Observation: raw, Observer: commontypes.OracleID(1)},
+			{Observation: raw, Observer: commontypes.OracleID(2)},
+		}
+	}
+
+	runRound := func(t *testing.T, queryID, obsID *pbtypes.Id) (*pbtypes.Outcome, []ocr3types.ReportPlus[[]byte], *forgeCapability) {
+		t.Helper()
+		cap := &forgeCapability{}
+		rp, err := NewReportingPlugin(
+			requests.NewStore[*ReportRequest](),
+			cap,
+			defaultBatchSize,
+			ocr3types.ReportingPluginConfig{F: 1},
+			defaultLimits(),
+			logger.Test(t),
+		)
+		require.NoError(t, err)
+
+		queryBytes, err := proto.Marshal(&pbtypes.Query{Ids: []*pbtypes.Id{queryID}})
+		require.NoError(t, err)
+
+		outcomeBytes, err := rp.Outcome(t.Context(), ocr3types.OutcomeContext{}, queryBytes, mkObs(obsID))
+		require.NoError(t, err)
+
+		outcome := &pbtypes.Outcome{}
+		require.NoError(t, proto.Unmarshal(outcomeBytes, outcome))
+
+		reports, err := rp.Reports(t.Context(), 1, outcomeBytes)
+		require.NoError(t, err)
+
+		return outcome, reports, cap
+	}
+
+	attackerID := &pbtypes.Id{
+		WorkflowExecutionId: sharedExecID,
+		WorkflowId:          attackerWfID,
+		WorkflowOwner:       attackerWfOwner,
+		WorkflowName:        attackerWfName,
+		ReportId:            attackerReport,
+		KeyId:               attackerKey,
+	}
+	victimID := &pbtypes.Id{
+		WorkflowExecutionId: sharedExecID,
+		WorkflowId:          victimWfID,
+		WorkflowOwner:       victimWfOwner,
+		WorkflowName:        victimWfName,
+		ReportId:            victimReport,
+		KeyId:               victimKey,
+	}
+
+	t.Run("honest leader: query identity matches observation identity", func(t *testing.T) {
+		outcome, _, cap := runRound(t, attackerID, attackerID)
+
+		require.Len(t, outcome.CurrentReports, 1)
+		rpt := outcome.CurrentReports[0]
+		require.Equal(t, attackerWfID, rpt.Id.WorkflowId)
+		require.Equal(t, attackerWfOwner, rpt.Id.WorkflowOwner)
+		require.Equal(t, attackerKey, rpt.Id.KeyId)
+		require.Equal(t, attackerWfID, cap.aggWfID, "aggregator selected for attacker's workflow")
+		require.Equal(t, attackerWfID, cap.encWfID, "encoder selected for attacker's workflow")
+	})
+
+	t.Run("byzantine leader: query carries victim identity, observations carry attacker identity", func(t *testing.T) {
+		outcome, _, _ := runRound(t, victimID, attackerID)
+
+		require.Empty(t, outcome.CurrentReports,
+			"forged report must not be produced when observation identity does not match query identity")
+	})
+}

pkg/capabilities/consensus/ocr3/reporting_plugin.go

@@ -194,9 +194,50 @@ func (r *reportingPlugin) Observation(ctx context.Context, outctx ocr3types.Outc
 }
 
 func (r *reportingPlugin) ValidateObservation(ctx context.Context, outctx ocr3types.OutcomeContext, query types.Query, ao types.AttributedObservation) error {
+	q := &pbtypes.Query{}
+	if err := proto.Unmarshal(query, q); err != nil {
+		return err
+	}
+
+	obs := &pbtypes.Observations{}
+	if err := proto.Unmarshal(ao.Observation, obs); err != nil {
+		return err
+	}
+
+	for _, request := range obs.Observations {
+		if request == nil || request.Id == nil {
+			continue
+		}
+		for _, qid := range q.Ids {
+			if qid == nil {
+				continue
+			}
+			if request.Id.WorkflowExecutionId != qid.WorkflowExecutionId {
+				continue
+			}
+			if !idMatches(request.Id, qid) {
+				return fmt.Errorf("observation identity does not match query identity for execution ID %s: observation workflow ID %s, query workflow ID %s",
+					request.Id.WorkflowExecutionId, request.Id.WorkflowId, qid.WorkflowId)
+			}
+			break
+		}
+	}
 	return nil
 }
 
+func idMatches(a, b *pbtypes.Id) bool {
+	if a == nil || b == nil {
+		return false
+	}
+	return a.WorkflowId == b.WorkflowId &&
+		a.WorkflowOwner == b.WorkflowOwner &&
+		a.WorkflowName == b.WorkflowName &&
+		a.ReportId == b.ReportId &&
+		a.KeyId == b.KeyId &&
+		a.WorkflowDonId == b.WorkflowDonId &&
+		a.WorkflowDonConfigVersion == b.WorkflowDonConfigVersion
+}
+
 func (r *reportingPlugin) ObservationQuorum(ctx context.Context, outctx ocr3types.OutcomeContext, query types.Query, aos []types.AttributedObservation) (bool, error) {
 	return quorumhelper.ObservationCountReachesObservationQuorum(quorumhelper.QuorumTwoFPlusOne, r.config.N, r.config.F, aos), nil
 }
@@ -229,6 +270,8 @@ type encoderConfig struct {
 func (r *reportingPlugin) Outcome(ctx context.Context, outctx ocr3types.OutcomeContext, query types.Query, attributedObservations []types.AttributedObservation) (ocr3types.Outcome, error) {
 	// execution ID -> oracle ID -> list of observations
 	execIDToOracleObservations := map[string]map[ocrcommon.OracleID][]values.Value{}
+	// execution ID -> oracle ID -> observation identity
+	execIDToObsIDs := map[string]map[ocrcommon.OracleID]*pbtypes.Id{}
 	seenWorkflowIDs := map[string]int{}
 	var sortedTimestamps []*timestamppb.Timestamp
 	var finalTimestamp *timestamppb.Timestamp
@@ -282,10 +325,15 @@ func (r *reportingPlugin) Outcome(ctx context.Context, outctx ocr3types.OutcomeC
 				continue
 			}
 
-			if _, ok := execIDToOracleObservations[weid]; !ok {
-				execIDToOracleObservations[weid] = make(map[ocrcommon.OracleID][]values.Value)
-			}
-			execIDToOracleObservations[weid][attributedObservation.Observer] = obsList.Underlying
+		if _, ok := execIDToOracleObservations[weid]; !ok {
+			execIDToOracleObservations[weid] = make(map[ocrcommon.OracleID][]values.Value)
+		}
+		execIDToOracleObservations[weid][attributedObservation.Observer] = obsList.Underlying
+
+		if _, ok := execIDToObsIDs[weid]; !ok {
+			execIDToObsIDs[weid] = make(map[ocrcommon.OracleID]*pbtypes.Id)
+		}
+		execIDToObsIDs[weid][attributedObservation.Observer] = request.Id
 
 			sha, err := shaForOverriddenEncoder(request)
 			if err != nil {
@@ -364,8 +412,19 @@ func (r *reportingPlugin) Outcome(ctx context.Context, outctx ocr3types.OutcomeC
 			lggr.Debugw("could not find existing outcome for workflow, aggregator will create a new one")
 		}
 
-		if len(obs) < (2*r.config.F + 1) {
-			lggr.Debugw("insufficient observations for workflow execution id")
+		matchingObs := make(map[ocrcommon.OracleID][]values.Value)
+		if obsIDs, ok := execIDToObsIDs[weid.WorkflowExecutionId]; ok {
+			for observer, obsValues := range obs {
+				if obsID, ok := obsIDs[observer]; ok && idMatches(obsID, weid) {
+					matchingObs[observer] = obsValues
+				}
+			}
+		} else {
+			matchingObs = obs
+		}
+
+		if len(matchingObs) < (2*r.config.F + 1) {
+			lggr.Debugw("insufficient observations with matching identity for workflow execution id")
 			continue
 		}
 
@@ -375,7 +434,7 @@ func (r *reportingPlugin) Outcome(ctx context.Context, outctx ocr3types.OutcomeC
 			continue
 		}
 
-		outcome, err2 := agg.Aggregate(lggr, workflowOutcome, obs, r.config.F)
+		outcome, err2 := agg.Aggregate(lggr, workflowOutcome, matchingObs, r.config.F)
 		if err2 != nil {
 			lggr.Errorw("error aggregating outcome", "error", err2)
 			continue

pkg/capabilities/consensus/ocr3/reporting_plugin_test.go

@@ -723,24 +723,23 @@ func TestReportingPlugin_Outcome_ShouldPruneOldOutcomes(t *testing.T) {
 	rp, err := NewReportingPlugin(s, mcap, defaultBatchSize, ocr3types.ReportingPluginConfig{}, defaultLimits(), lggr)
 	require.NoError(t, err)
 
-	weid := uuid.New().String()
 	wowner := uuid.New().String()
 	id := &pbtypes.Id{
-		WorkflowExecutionId: weid,
+		WorkflowExecutionId: uuid.New().String(),
 		WorkflowId:          workflowTestID,
 		WorkflowOwner:       wowner,
 		WorkflowName:        workflowTestName,
 		ReportId:            reportTestID,
 	}
 	id2 := &pbtypes.Id{
-		WorkflowExecutionId: weid,
+		WorkflowExecutionId: uuid.New().String(),
 		WorkflowId:          workflowTestID2,
 		WorkflowOwner:       wowner,
 		WorkflowName:        workflowTestName,
 		ReportId:            reportTestID,
 	}
 	id3 := &pbtypes.Id{
-		WorkflowExecutionId: weid,
+		WorkflowExecutionId: uuid.New().String(),
 		WorkflowId:          workflowTestID3,
 		WorkflowOwner:       wowner,
 		WorkflowName:        workflowTestName,
@@ -841,6 +840,8 @@ func TestReportPlugin_Outcome_ShouldReturnMedianTimestamp(t *testing.T) {
 	require.NoError(t, err)
 
 	weid := uuid.New().String()
+	weid2 := uuid.New().String()
+	weid3 := uuid.New().String()
 	wowner := uuid.New().String()
 	id := &pbtypes.Id{
 		WorkflowExecutionId: weid,
@@ -850,14 +851,14 @@ func TestReportPlugin_Outcome_ShouldReturnMedianTimestamp(t *testing.T) {
 		ReportId:            reportTestID,
 	}
 	id2 := &pbtypes.Id{
-		WorkflowExecutionId: weid,
+		WorkflowExecutionId: weid2,
 		WorkflowId:          workflowTestID2,
 		WorkflowOwner:       wowner,
 		WorkflowName:        workflowTestName,
 		ReportId:            reportTestID,
 	}
 	id3 := &pbtypes.Id{
-		WorkflowExecutionId: weid,
+		WorkflowExecutionId: weid3,
 		WorkflowId:          workflowTestID3,
 		WorkflowOwner:       wowner,
 		WorkflowName:        workflowTestName,