Validate report context and ReportProcessed events in the Stellar write path

chain_capabilities/stellar/actions/forwarder_client.go

@@ -291,6 +291,11 @@ func (fc *forwarderClient) GetReportProcessedEvents(
 		}
 
 		for i, e := range resp.Events {
+			if reason := reportProcessedEventMismatch(e, fc.forwarderAddress, searchRange); reason != "" {
+				fc.lggr.Warnw("Ignoring ReportProcessed event that does not match the query",
+					append([]any{"reason", reason, "txHash", e.TransactionHash, "ledger", e.Ledger, "contractID", e.ContractID, "eventType", e.EventType}, transmissionID.LogAttrs()...)...)
+				continue
+			}
 			if e.TransactionHash == "" {
 				return nil, fmt.Errorf("empty tx hash at event index %d", i)
 			}
@@ -313,6 +318,20 @@ func (fc *forwarderClient) GetReportProcessedEvents(
 	return nil, fmt.Errorf("too many ReportProcessed event pages for range %d-%d", searchRange.StartLedger, searchRange.EndLedger)
 }
 
+// reportProcessedEventMismatch returns why an event does not belong to the query, or "" when it does.
+func reportProcessedEventMismatch(e stellartypes.EventInfo, forwarderAddress string, searchRange EventSearchRange) string {
+	if e.EventType != stellartypes.EventTypeContract {
+		return "not a contract event"
+	}
+	if e.ContractID != forwarderAddress {
+		return "contract id does not match forwarder"
+	}
+	if e.Ledger < searchRange.StartLedger || e.Ledger > searchRange.EndLedger {
+		return "ledger is outside the search range"
+	}
+	return ""
+}
+
 func (fc *forwarderClient) GetReportProcessedEventSearchRange(ctx context.Context) (EventSearchRange, error) {
 	endLedger, err := fc.GetReportProcessedEventSearchEndLedger(ctx)
 	if err != nil {

chain_capabilities/stellar/actions/forwarder_client_test.go

@@ -87,6 +87,25 @@ func TestForwarderClient_InvokeOnReport(t *testing.T) {
 	})
 }
 
+// reportProcessedEvent builds an EventInfo as the forwarder emits it for transmissionID.
+func reportProcessedEvent(t *testing.T, transmissionID TransmissionID, txHash string, ledger uint32, success bool) stellartypes.EventInfo {
+	t.Helper()
+	filter, err := NewCREForwarderCodec().EncodeReportProcessedTopicFilter(transmissionID)
+	require.NoError(t, err)
+	topics := make([]stellartypes.ScVal, len(filter.Segments))
+	for i, segment := range filter.Segments {
+		topics[i] = *segment.Value
+	}
+	return stellartypes.EventInfo{
+		EventType:       stellartypes.EventTypeContract,
+		Ledger:          ledger,
+		ContractID:      testForwarderAddress,
+		TransactionHash: txHash,
+		Topics:          topics,
+		Value:           stellartypes.ScVal{Type: stellartypes.ScValTypeBool, Bool: &success},
+	}
+}
+
 func TestForwarderClient_GetReportProcessedEvents(t *testing.T) {
 	t.Parallel()
 	lggr := logger.Test(t)
@@ -95,32 +114,77 @@ func TestForwarderClient_GetReportProcessedEvents(t *testing.T) {
 	require.NoError(t, err)
 	searchRange := EventSearchRange{StartLedger: 100, EndLedger: 200}
 
+	expectEvents := func(t *testing.T, events ...stellartypes.EventInfo) CREForwarderClient {
+		t.Helper()
+		svc := mocks.NewStellarService(t)
+		svc.EXPECT().GetEvents(mock.Anything, mock.Anything).
+			Return(stellartypes.GetEventsResponse{Events: events}, nil).Once()
+		return newForwarderClient(svc, lggr, testForwarderAddress, 100)
+	}
+
 	t.Run("happy path", func(t *testing.T) {
 		t.Parallel()
 		svc := mocks.NewStellarService(t)
-		success := true
 		svc.EXPECT().GetEvents(mock.Anything, mock.MatchedBy(func(req stellartypes.GetEventsRequest) bool {
 			return req.StartLedger == searchRange.StartLedger &&
 				req.EndLedger == searchRange.EndLedger &&
 				req.Pagination != nil &&
 				req.Pagination.Limit == reportProcessedEventPageLimit
 		})).
 			Return(stellartypes.GetEventsResponse{
-				Events: []stellartypes.EventInfo{{
-					TransactionHash: testTxHash,
-					Ledger:          150,
-					Value:           stellartypes.ScVal{Type: stellartypes.ScValTypeBool, Bool: &success},
-				}},
+				Events: []stellartypes.EventInfo{reportProcessedEvent(t, transmissionID, testTxHash, 150, true)},
 			}, nil).Once()
 		client := newForwarderClient(svc, lggr, testForwarderAddress, 100)
 
 		events, err := client.GetReportProcessedEvents(t.Context(), transmissionID, searchRange)
 		require.NoError(t, err)
 		require.Len(t, events, 1)
 		require.Equal(t, testTxHash, events[0].TxHash)
+		require.Equal(t, uint32(150), events[0].Ledger)
 		require.True(t, events[0].Success)
 	})
 
+	t.Run("skips diagnostic and system events", func(t *testing.T) {
+		t.Parallel()
+		system := reportProcessedEvent(t, transmissionID, "system", 150, true)
+		system.EventType = stellartypes.EventTypeSystem
+		diagnostic := reportProcessedEvent(t, transmissionID, "diagnostic-copy", 151, true)
+		diagnostic.EventType = stellartypes.EventType(99)
+		client := expectEvents(t, system, diagnostic, reportProcessedEvent(t, transmissionID, testTxHash, 152, false))
+
+		events, err := client.GetReportProcessedEvents(t.Context(), transmissionID, searchRange)
+		require.NoError(t, err)
+		require.Len(t, events, 1)
+		require.Equal(t, testTxHash, events[0].TxHash)
+		require.False(t, events[0].Success)
+	})
+
+	t.Run("skips events from another contract", func(t *testing.T) {
+		t.Parallel()
+		foreign := reportProcessedEvent(t, transmissionID, "other-contract", 150, true)
+		foreign.ContractID = testReceiverAddress
+		client := expectEvents(t, foreign)
+
+		events, err := client.GetReportProcessedEvents(t.Context(), transmissionID, searchRange)
+		require.NoError(t, err)
+		require.Empty(t, events)
+	})
+
+	t.Run("skips events outside the search range", func(t *testing.T) {
+		t.Parallel()
+		client := expectEvents(t,
+			reportProcessedEvent(t, transmissionID, "ledger-zero", 0, false),
+			reportProcessedEvent(t, transmissionID, "too-old", 99, false),
+			reportProcessedEvent(t, transmissionID, "too-new", 201, false),
+			reportProcessedEvent(t, transmissionID, testTxHash, 100, false),
+		)
+
+		events, err := client.GetReportProcessedEvents(t.Context(), transmissionID, searchRange)
+		require.NoError(t, err)
+		require.Len(t, events, 1)
+		require.Equal(t, testTxHash, events[0].TxHash)
+	})
+
 	t.Run("search range clamps to ledger 1 when history is short", func(t *testing.T) {
 		t.Parallel()
 		svc := mocks.NewStellarService(t)
@@ -136,16 +200,10 @@ func TestForwarderClient_GetReportProcessedEvents(t *testing.T) {
 	t.Run("drains paginated results", func(t *testing.T) {
 		t.Parallel()
 		svc := mocks.NewStellarService(t)
-		failed := false
-		success := true
 		svc.EXPECT().GetEvents(mock.Anything, mock.MatchedBy(func(req stellartypes.GetEventsRequest) bool {
 			return req.Pagination != nil && req.Pagination.Cursor == ""
 		})).Return(stellartypes.GetEventsResponse{
-			Events: []stellartypes.EventInfo{{
-				TransactionHash: "failed",
-				Ledger:          150,
-				Value:           stellartypes.ScVal{Type: stellartypes.ScValTypeBool, Bool: &failed},
-			}},
+			Events: []stellartypes.EventInfo{reportProcessedEvent(t, transmissionID, "failed", 150, false)},
 			Cursor: "next",
 		}, nil).Once()
 		svc.EXPECT().GetEvents(mock.Anything, mock.MatchedBy(func(req stellartypes.GetEventsRequest) bool {
@@ -154,11 +212,7 @@ func TestForwarderClient_GetReportProcessedEvents(t *testing.T) {
 				req.Pagination != nil &&
 				req.Pagination.Cursor == "next"
 		})).Return(stellartypes.GetEventsResponse{
-			Events: []stellartypes.EventInfo{{
-				TransactionHash: testTxHash,
-				Ledger:          151,
-				Value:           stellartypes.ScVal{Type: stellartypes.ScValTypeBool, Bool: &success},
-			}},
+			Events: []stellartypes.EventInfo{reportProcessedEvent(t, transmissionID, testTxHash, 151, true)},
 		}, nil).Once()
 		client := newForwarderClient(svc, lggr, testForwarderAddress, 100)
 
@@ -184,16 +238,7 @@ func TestForwarderClient_GetReportProcessedEvents(t *testing.T) {
 
 	t.Run("empty tx hash in event", func(t *testing.T) {
 		t.Parallel()
-		svc := mocks.NewStellarService(t)
-		success := true
-		svc.EXPECT().GetEvents(mock.Anything, mock.Anything).
-			Return(stellartypes.GetEventsResponse{
-				Events: []stellartypes.EventInfo{{
-					TransactionHash: "",
-					Value:           stellartypes.ScVal{Type: stellartypes.ScValTypeBool, Bool: &success},
-				}},
-			}, nil).Once()
-		client := newForwarderClient(svc, lggr, testForwarderAddress, 100)
+		client := expectEvents(t, reportProcessedEvent(t, transmissionID, "", 150, true))
 
 		_, err := client.GetReportProcessedEvents(t.Context(), transmissionID, searchRange)
 		require.Error(t, err)
@@ -202,15 +247,9 @@ func TestForwarderClient_GetReportProcessedEvents(t *testing.T) {
 
 	t.Run("non-bool event value", func(t *testing.T) {
 		t.Parallel()
-		svc := mocks.NewStellarService(t)
-		svc.EXPECT().GetEvents(mock.Anything, mock.Anything).
-			Return(stellartypes.GetEventsResponse{
-				Events: []stellartypes.EventInfo{{
-					TransactionHash: testTxHash,
-					Value:           stellartypes.ScVal{Type: stellartypes.ScValTypeU32},
-				}},
-			}, nil).Once()
-		client := newForwarderClient(svc, lggr, testForwarderAddress, 100)
+		event := reportProcessedEvent(t, transmissionID, testTxHash, 150, true)
+		event.Value = stellartypes.ScVal{Type: stellartypes.ScValTypeU32}
+		client := expectEvents(t, event)
 
 		_, err := client.GetReportProcessedEvents(t.Context(), transmissionID, searchRange)
 		require.Error(t, err)

chain_capabilities/stellar/actions/write_report.go

@@ -1,6 +1,7 @@
 package actions
 
 import (
+	"bytes"
 	"context"
 	"encoding/hex"
 	"errors"
@@ -13,6 +14,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/report"
 	caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors"
 	stellarcap "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/stellar"
 	commoncfg "github.com/smartcontractkit/chainlink-common/pkg/config"
@@ -21,6 +23,8 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
 	stellartypes "github.com/smartcontractkit/chainlink-common/pkg/types/chains/stellar"
+	"github.com/smartcontractkit/chainlink-protos/cre/go/sdk"
+	libocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types"
 
 	capcommon "github.com/smartcontractkit/capabilities/chain_capabilities/common"
 	ts "github.com/smartcontractkit/capabilities/chain_capabilities/common/transmission_schedule"
@@ -360,6 +364,21 @@ func (s *Stellar) validateWriteReportInputs(metadata capabilities.RequestMetadat
 	if reportMetadata.WorkflowID != metadata.WorkflowID {
 		return fmt.Errorf("%s report workflowID does not match request metadata", capcommon.UserError)
 	}
+	return validateReportContext(request.Report)
+}
+
+// validateReportContext checks that the report context the forwarder verifies signatures
+// against is the one consensus derived from this report's config digest and sequence number,
+// rather than caller-chosen bytes.
+func validateReportContext(signedReport *sdk.ReportResponse) error {
+	var configDigest libocrtypes.ConfigDigest
+	if len(signedReport.ConfigDigest) != len(configDigest) {
+		return fmt.Errorf("%s config digest has invalid length: got %d, want %d", capcommon.UserError, len(signedReport.ConfigDigest), len(configDigest))
+	}
+	copy(configDigest[:], signedReport.ConfigDigest)
+	if !bytes.Equal(signedReport.ReportContext, report.GenerateReportContext(signedReport.SeqNr, configDigest)) {
+		return fmt.Errorf("%s report context does not match config digest and sequence number", capcommon.UserError)
+	}
 	return nil
 }
 

chain_capabilities/stellar/actions/write_report_test.go

@@ -1,6 +1,7 @@
 package actions
 
 import (
+	"bytes"
 	"context"
 	"encoding/hex"
 	"errors"
@@ -15,12 +16,14 @@ import (
 
 	"google.golang.org/protobuf/proto"
 
+	libocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types"
 	p2ptypes "github.com/smartcontractkit/libocr/ragep2p/types"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	ocrtypes "github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/ocr3/types"
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/report"
 	stellarcap "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/stellar"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
@@ -131,15 +134,26 @@ func newWRReportFixture(t *testing.T) (ocrtypes.Metadata, capabilities.RequestMe
 	}
 	req := &stellarcap.WriteReportRequest{
 		ContractId: testReceiverAddress,
-		Report: &workflowpb.ReportResponse{
-			RawReport:     encoded,
-			ReportContext: make([]byte, ocrReportContextLen),
-			Sigs:          wrTestSigs(),
-		},
+		Report:     wrSignedReport(encoded),
 	}
 	return rm, reqMeta, req
 }
 
+// wrSignedReport wraps rawReport with a config digest, sequence number and the report
+// context consensus derives from them, plus test signatures.
+func wrSignedReport(rawReport []byte) *workflowpb.ReportResponse {
+	var configDigest libocrtypes.ConfigDigest
+	copy(configDigest[:], commontest.RandomBytes(32))
+	seqNr := uint64(42)
+	return &workflowpb.ReportResponse{
+		RawReport:     rawReport,
+		ConfigDigest:  configDigest[:],
+		SeqNr:         seqNr,
+		ReportContext: report.GenerateReportContext(seqNr, configDigest),
+		Sigs:          wrTestSigs(),
+	}
+}
+
 func wrTestSigs() []*workflowpb.AttributedSignature {
 	// ed25519 OCR sigs: 32-byte pubkey || 64-byte signature. Distinct leading
 	// pubkey bytes so the codec's ascending-by-pubkey ordering is well-defined.
@@ -297,7 +311,9 @@ func reportProcessedEventsForFixture(t *testing.T, rm ocrtypes.Metadata, receive
 
 	return stellartypes.GetEventsResponse{
 		Events: []stellartypes.EventInfo{{
+			EventType:       stellartypes.EventTypeContract,
 			Ledger:          100,
+			ContractID:      testForwarderAddress,
 			TransactionHash: testTxHash,
 			Topics: []stellartypes.ScVal{
 				{Type: stellartypes.ScValTypeSymbol, Symbol: &eventName},
@@ -478,11 +494,7 @@ func TestWriteReport_Validation(t *testing.T) {
 		_, reqMeta, _ := newWRReportFixture(t)
 		req := &stellarcap.WriteReportRequest{
 			ContractId: testReceiverAddress,
-			Report: &workflowpb.ReportResponse{
-				RawReport:     []byte("garbage"),
-				ReportContext: make([]byte, ocrReportContextLen),
-				Sigs:          wrTestSigs(),
-			},
+			Report:     wrSignedReport([]byte("garbage")),
 		}
 
 		_, err := h.stellar.WriteReport(t.Context(), reqMeta, req)
@@ -534,6 +546,39 @@ func TestWriteReport_Validation(t *testing.T) {
 		require.Contains(t, err.Error(), "workflowID does not match")
 	})
 
+	t.Run("report context not derived from config digest and seqNr", func(t *testing.T) {
+		t.Parallel()
+		h := newWriteReportHelper(t)
+		_, reqMeta, req := newWRReportFixture(t)
+		req.Report.ReportContext = bytes.Repeat([]byte{0xFF}, ocrReportContextLen)
+
+		_, err := h.stellar.WriteReport(t.Context(), reqMeta, req)
+		require.NotNil(t, err)
+		require.Contains(t, err.Error(), "report context does not match config digest and sequence number")
+	})
+
+	t.Run("report context from a different seqNr", func(t *testing.T) {
+		t.Parallel()
+		h := newWriteReportHelper(t)
+		_, reqMeta, req := newWRReportFixture(t)
+		req.Report.SeqNr++
+
+		_, err := h.stellar.WriteReport(t.Context(), reqMeta, req)
+		require.NotNil(t, err)
+		require.Contains(t, err.Error(), "report context does not match config digest and sequence number")
+	})
+
+	t.Run("invalid config digest length", func(t *testing.T) {
+		t.Parallel()
+		h := newWriteReportHelper(t)
+		_, reqMeta, req := newWRReportFixture(t)
+		req.Report.ConfigDigest = nil
+
+		_, err := h.stellar.WriteReport(t.Context(), reqMeta, req)
+		require.NotNil(t, err)
+		require.Contains(t, err.Error(), "config digest has invalid length")
+	})
+
 	t.Run("report size exceeds limit", func(t *testing.T) {
 		t.Parallel()
 		h := newWriteReportHelper(t)
@@ -543,11 +588,7 @@ func TestWriteReport_Validation(t *testing.T) {
 
 		req := &stellarcap.WriteReportRequest{
 			ContractId: testReceiverAddress,
-			Report: &workflowpb.ReportResponse{
-				RawReport:     append(encoded, make([]byte, 20_000)...),
-				ReportContext: make([]byte, ocrReportContextLen),
-				Sigs:          wrTestSigs(),
-			},
+			Report:     wrSignedReport(append(encoded, make([]byte, 20_000)...)),
 		}
 
 		_, capErr := h.stellar.WriteReport(t.Context(), reqMeta, req)