From 6e152930772739da8525ba0541df81edec58bec2 Mon Sep 17 00:00:00 2001 From: mchain0 Date: Tue, 4 Aug 2026 15:54:37 +0200 Subject: [PATCH] cre-5770: extra observations validation for reporting --- .../consensus/ocr3/identity_forge_test.go | 162 ++++++++++++++++++ .../consensus/ocr3/reporting_plugin.go | 73 +++++++- .../consensus/ocr3/reporting_plugin_test.go | 13 +- 3 files changed, 235 insertions(+), 13 deletions(-) create mode 100644 pkg/capabilities/consensus/ocr3/identity_forge_test.go diff --git a/pkg/capabilities/consensus/ocr3/identity_forge_test.go b/pkg/capabilities/consensus/ocr3/identity_forge_test.go new file mode 100644 index 0000000000..2b0ff20ae0 --- /dev/null +++ b/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") + }) +} diff --git a/pkg/capabilities/consensus/ocr3/reporting_plugin.go b/pkg/capabilities/consensus/ocr3/reporting_plugin.go index cc0e2688a8..dfa769c187 100644 --- a/pkg/capabilities/consensus/ocr3/reporting_plugin.go +++ b/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 diff --git a/pkg/capabilities/consensus/ocr3/reporting_plugin_test.go b/pkg/capabilities/consensus/ocr3/reporting_plugin_test.go index 9a99b5a55c..f793597a35 100644 --- a/pkg/capabilities/consensus/ocr3/reporting_plugin_test.go +++ b/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,