Make plugin robust against outcome manipulation (#799)

## Motivation The OCR plugin fails to deduplicate the values of slices in each oracle’s reported observation. This allows a single node to introduce duplicate values in its observation slices, causing the plugin to erroneously trust that certain values are the consensus among the DON. ## Solution Extending observation validation to include deduplication validation.
smartcontractkit · May 3, 2024 · 9686b07 · 9686b07
1 parent c8385ba
commit 9686b07
Show file tree

Hide file tree

Showing 3 changed files with 188 additions and 3 deletions.
diff --git a/.changeset/ten-adults-try.md b/.changeset/ten-adults-try.md
@@ -0,0 +1,5 @@
+---
+"ccip": patch
+---
+
+Extending observation validation to include deduplication validation, in order to avoid from observation manipulation
diff --git a/core/services/ocr2/plugins/liquiditymanager/plugin.go b/core/services/ocr2/plugins/liquiditymanager/plugin.go
@@ -146,12 +146,31 @@ func (p *Plugin) ValidateObservation(outctx ocr3types.OutcomeContext, query ocrt
 	// todo: improve logging - including duration of each phase, etc.
 	p.lggr.Infow("in validate observation", "seqNr", outctx.SeqNr, "phase", "ValidateObservation")
 
-	_, err := models.DecodeObservation(ao.Observation)
+	obs, err := models.DecodeObservation(ao.Observation)
 	if err != nil {
 		return fmt.Errorf("invalid observation: %w", err)
 	}
 
-	// todo: consider adding more validations
+	if err := validateDedupedItems(func(liq models.NetworkLiquidity) string {
+		return fmt.Sprintf("%d", liq.Network)
+	}, obs.LiquidityPerChain...); err != nil {
+		return fmt.Errorf("invalid LiquidityPerChain: %w", err)
+	}
+	if err := validateDedupedItems(dedupKeyObject, obs.ResolvedTransfers...); err != nil {
+		return fmt.Errorf("invalid ResolvedTransfers: %w", err)
+	}
+	if err := validateDedupedItems(dedupKeyObject, obs.PendingTransfers...); err != nil {
+		return fmt.Errorf("invalid PendingTransfers: %w", err)
+	}
+	if err := validateDedupedItems(dedupKeyObject, obs.InflightTransfers...); err != nil {
+		return fmt.Errorf("invalid InflightTransfers: %w", err)
+	}
+	if err := validateDedupedItems(dedupKeyObject, obs.Edges...); err != nil {
+		return fmt.Errorf("invalid Edges: %w", err)
+	}
+	if err := validateDedupedItems(dedupKeyObject, obs.ConfigDigests...); err != nil {
+		return fmt.Errorf("invalid ConfigDigests: %w", err)
+	}
 
 	return nil
 }
@@ -716,3 +735,20 @@ func (p *Plugin) resolveProposedTransfers(ctx context.Context, lggr logger.Logge
 
 	return resolvedTransfers, nil
 }
+
+func dedupKeyObject[T any](item T) string {
+	return fmt.Sprintf("%+v", item)
+}
+
+// validateDedupedItems checks if there are any duplicated items in the provided slice.
+func validateDedupedItems[T any](keyFn func(T) string, items ...T) error {
+	existing := map[string]bool{}
+	for _, item := range items {
+		k := keyFn(item)
+		if existing[k] {
+			return fmt.Errorf("duplicated item")
+		}
+		existing[k] = true
+	}
+	return nil
+}
diff --git a/core/services/ocr2/plugins/liquiditymanager/plugin_test.go b/core/services/ocr2/plugins/liquiditymanager/plugin_test.go
@@ -328,13 +328,98 @@ func TestPlugin_ValidateObservation(t *testing.T) {
 			name: "some observation",
 			obs: models.NewObservation(
 				[]models.NetworkLiquidity{},
-				[]models.Transfer{{}, {}},
+				[]models.Transfer{{}},
 				[]models.PendingTransfer{},
 				[]models.Transfer{},
 				[]models.Edge{},
 				[]models.ConfigDigestWithMeta{},
 			).Encode(),
 		},
+
+		{
+			name: "deduped liquidity observations",
+			obs: models.NewObservation(
+				[]models.NetworkLiquidity{{Network: 1, Liquidity: ubig.New(big.NewInt(1))}, {Network: 1, Liquidity: ubig.New(big.NewInt(2))}},
+				[]models.Transfer{},
+				[]models.PendingTransfer{},
+				[]models.Transfer{},
+				[]models.Edge{},
+				[]models.ConfigDigestWithMeta{},
+			).Encode(),
+			expErr: func(t *testing.T, err error) {
+				assert.Error(t, err)
+			},
+		},
+		{
+			name: "deduped resolved transfers",
+			obs: models.NewObservation(
+				[]models.NetworkLiquidity{},
+				[]models.Transfer{{From: 1}, {From: 1}},
+				[]models.PendingTransfer{},
+				[]models.Transfer{},
+				[]models.Edge{},
+				[]models.ConfigDigestWithMeta{},
+			).Encode(),
+			expErr: func(t *testing.T, err error) {
+				assert.Error(t, err)
+			},
+		},
+		{
+			name: "deduped pending transfers",
+			obs: models.NewObservation(
+				[]models.NetworkLiquidity{},
+				[]models.Transfer{},
+				[]models.PendingTransfer{{ID: "1"}, {ID: "1"}},
+				[]models.Transfer{},
+				[]models.Edge{},
+				[]models.ConfigDigestWithMeta{},
+			).Encode(),
+			expErr: func(t *testing.T, err error) {
+				assert.Error(t, err)
+			},
+		},
+		{
+			name: "deduped inflight transfers",
+			obs: models.NewObservation(
+				[]models.NetworkLiquidity{},
+				[]models.Transfer{},
+				[]models.PendingTransfer{},
+				[]models.Transfer{{From: 1}, {From: 1}},
+				[]models.Edge{},
+				[]models.ConfigDigestWithMeta{},
+			).Encode(),
+			expErr: func(t *testing.T, err error) {
+				assert.Error(t, err)
+			},
+		},
+		{
+			name: "deduped edges",
+			obs: models.NewObservation(
+				[]models.NetworkLiquidity{},
+				[]models.Transfer{},
+				[]models.PendingTransfer{},
+				[]models.Transfer{},
+				[]models.Edge{{Source: 1, Dest: 2}, {Source: 1, Dest: 2}},
+				[]models.ConfigDigestWithMeta{},
+			).Encode(),
+			expErr: func(t *testing.T, err error) {
+				assert.Error(t, err)
+			},
+		},
+		{
+			name: "deduped config digest",
+			obs: models.NewObservation(
+				[]models.NetworkLiquidity{},
+				[]models.Transfer{},
+				[]models.PendingTransfer{},
+				[]models.Transfer{},
+				[]models.Edge{},
+				[]models.ConfigDigestWithMeta{{NetworkSel: 1}, {NetworkSel: 1}},
+			).Encode(),
+			expErr: func(t *testing.T, err error) {
+				assert.Error(t, err)
+			},
+		},
 	}
 
 	for _, tc := range testCases {
@@ -354,6 +439,65 @@ func TestPlugin_ValidateObservation(t *testing.T) {
 	}
 }
 
+func Test_validateDedupedItems(t *testing.T) {
+	tests := []struct {
+		name    string
+		keyFn   func(*models.Transfer) string
+		items   []*models.Transfer
+		wantErr bool
+	}{
+		{
+			name: "no duplicates",
+			items: []*models.Transfer{
+				{From: 1},
+				{From: 2},
+				{From: 3},
+			},
+			wantErr: false,
+		},
+		{
+			name: "duplicates",
+			items: []*models.Transfer{
+				{From: 1},
+				{From: 2},
+				{From: 1},
+			},
+			wantErr: true,
+		},
+		{
+			name:    "empty",
+			items:   []*models.Transfer{},
+			wantErr: false,
+		},
+		{
+			name: "custom keyFn",
+			keyFn: func(t *models.Transfer) string {
+				return fmt.Sprintf("%d", t.From)
+			},
+			items: []*models.Transfer{
+				{From: 1, To: 2},
+				{From: 1, To: 3},
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			keyFn := tc.keyFn
+			if keyFn == nil {
+				keyFn = dedupKeyObject
+			}
+			err := validateDedupedItems(keyFn, tc.items...)
+			if tc.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+		})
+	}
+}
+
 func TestPlugin_ObservationQuorum(t *testing.T) {
 	p := newPluginWithMocksAndDefaults(t)
 	res, err := p.plugin.ObservationQuorum(ocr3types.OutcomeContext{}, ocrtypes.Query{})