Merge branch 'develop' into make_grafana_token_optional

.changeset/eighty-hotels-sit.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+Fix panic if mercury server returns error #bugfix

.changeset/hungry-carpets-flow.md

@@ -0,0 +1,5 @@
+---
+"chainlink": minor
+---
+
+Added a mechanism to validate forwarders for OCR2 and fallback to EOA if necessary #added

CHANGELOG.md

@@ -189,7 +189,7 @@
   You may disable if this results in excessive log volume. Disable like so:
 
   ```
-  [Pipeline]
+  [JobPipeline]
   VerboseLogging = false
   ```
 
@@ -219,7 +219,7 @@
 
 - [#12404](https://github.com/smartcontractkit/chainlink/pull/12404) [`b74079b672`](https://github.com/smartcontractkit/chainlink/commit/b74079b672f36fb0c241f90ea1e875ea3a9524da) Thanks [@HenryNguyen5](https://github.com/HenryNguyen5)! - Add OCR3 capability contract wrapper
 
-- [#12498](https://github.com/smartcontractkit/chainlink/pull/12498) [`1c576d0e34`](https://github.com/smartcontractkit/chainlink/commit/1c576d0e34d93a6298ddcb662ee89fd04eeda53e) Thanks [@samsondav](https://github.com/samsondav)! - Add new config option Pipeline.VerboseLogging
+- [#12498](https://github.com/smartcontractkit/chainlink/pull/12498) [`1c576d0e34`](https://github.com/smartcontractkit/chainlink/commit/1c576d0e34d93a6298ddcb662ee89fd04eeda53e) Thanks [@samsondav](https://github.com/samsondav)! - Add new config option JobPipeline.VerboseLogging
 
   VerboseLogging enables detailed logging of pipeline execution steps. This is
   disabled by default because it increases log volume for pipeline runs, but can
@@ -230,7 +230,7 @@
   Set it like the following example:
 
   ```
-  [Pipeline]
+  [JobPipeline]
   VerboseLogging = true
   ```
 

common/txmgr/txmgr.go

@@ -47,6 +47,7 @@ type TxManager[
 	Trigger(addr ADDR)
 	CreateTransaction(ctx context.Context, txRequest txmgrtypes.TxRequest[ADDR, TX_HASH]) (etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error)
 	GetForwarderForEOA(eoa ADDR) (forwarder ADDR, err error)
+	GetForwarderForEOAOCR2Feeds(eoa, ocr2AggregatorID ADDR) (forwarder ADDR, err error)
 	RegisterResumeCallback(fn ResumeCallback)
 	SendNativeToken(ctx context.Context, chainID CHAIN_ID, from, to ADDR, value big.Int, gasLimit uint64) (etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error)
 	Reset(addr ADDR, abandon bool) error
@@ -553,6 +554,15 @@ func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) GetForward
 	return
 }
 
+// GetForwarderForEOAOCR2Feeds calls forwarderMgr to get a proper forwarder for a given EOA and checks if its set as a transmitter on the OCR2Aggregator contract.
+func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) GetForwarderForEOAOCR2Feeds(eoa, ocr2Aggregator ADDR) (forwarder ADDR, err error) {
+	if !b.txConfig.ForwardersEnabled() {
+		return forwarder, fmt.Errorf("forwarding is not enabled, to enable set Transactions.ForwardersEnabled =true")
+	}
+	forwarder, err = b.fwdMgr.ForwarderForOCR2Feeds(eoa, ocr2Aggregator)
+	return
+}
+
 func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) checkEnabled(ctx context.Context, addr ADDR) error {
 	if err := b.keyStore.CheckEnabled(ctx, addr, b.chainID); err != nil {
 		return fmt.Errorf("cannot send transaction from %s on chain ID %s: %w", addr, b.chainID.String(), err)
@@ -649,6 +659,10 @@ func (n *NullTxManager[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) Cre
 func (n *NullTxManager[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) GetForwarderForEOA(addr ADDR) (fwdr ADDR, err error) {
 	return fwdr, err
 }
+func (n *NullTxManager[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) GetForwarderForEOAOCR2Feeds(_, _ ADDR) (fwdr ADDR, err error) {
+	return fwdr, err
+}
+
 func (n *NullTxManager[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) Reset(addr ADDR, abandon bool) error {
 	return nil
 }

common/txmgr/types/forwarder_manager.go

@@ -9,6 +9,7 @@ import (
 type ForwarderManager[ADDR types.Hashable] interface {
 	services.Service
 	ForwarderFor(addr ADDR) (forwarder ADDR, err error)
+	ForwarderForOCR2Feeds(eoa, ocr2Aggregator ADDR) (forwarder ADDR, err error)
 	// Converts payload to be forwarder-friendly
 	ConvertPayload(dest ADDR, origPayload []byte) ([]byte, error)
 }

core/chains/evm/forwarders/forwarder_manager.go

@@ -2,13 +2,15 @@ package forwarders
 
 import (
 	"context"
+	"slices"
 	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	pkgerrors "github.com/pkg/errors"
+	"github.com/smartcontractkit/libocr/gethwrappers2/ocr2aggregator"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
@@ -131,6 +133,42 @@ func (f *FwdMgr) ForwarderFor(addr common.Address) (forwarder common.Address, er
 	return common.Address{}, pkgerrors.Errorf("Cannot find forwarder for given EOA")
 }
 
+func (f *FwdMgr) ForwarderForOCR2Feeds(eoa, ocr2Aggregator common.Address) (forwarder common.Address, err error) {
+	fwdrs, err := f.ORM.FindForwardersByChain(f.ctx, big.Big(*f.evmClient.ConfiguredChainID()))
+	if err != nil {
+		return common.Address{}, err
+	}
+
+	offchainAggregator, err := ocr2aggregator.NewOCR2Aggregator(ocr2Aggregator, f.evmClient)
+	if err != nil {
+		return common.Address{}, err
+	}
+
+	transmitters, err := offchainAggregator.GetTransmitters(&bind.CallOpts{Context: f.ctx})
+	if err != nil {
+		return common.Address{}, pkgerrors.Errorf("failed to get ocr2 aggregator transmitters: %s", err.Error())
+	}
+
+	for _, fwdr := range fwdrs {
+		if !slices.Contains(transmitters, fwdr.Address) {
+			f.logger.Criticalw("Forwarder is not set as a transmitter", "forwarder", fwdr.Address, "ocr2Aggregator", ocr2Aggregator, "err", err)
+			continue
+		}
+
+		eoas, err := f.getContractSenders(fwdr.Address)
+		if err != nil {
+			f.logger.Errorw("Failed to get forwarder senders", "forwarder", fwdr.Address, "err", err)
+			continue
+		}
+		for _, addr := range eoas {
+			if addr == eoa {
+				return fwdr.Address, nil
+			}
+		}
+	}
+	return common.Address{}, pkgerrors.Errorf("Cannot find forwarder for given EOA")
+}
+
 func (f *FwdMgr) ConvertPayload(dest common.Address, origPayload []byte) ([]byte, error) {
 	databytes, err := f.getForwardedPayload(dest, origPayload)
 	if err != nil {

core/chains/evm/forwarders/forwarder_manager_test.go

@@ -2,19 +2,23 @@ package forwarders_test
 
 import (
 	"math/big"
+	"slices"
 	"testing"
 	"time"
 
-	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
-
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind/backends"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/smartcontractkit/libocr/gethwrappers2/testocr2aggregator"
+
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils"
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/testhelpers"
 
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/client"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/forwarders"
@@ -150,3 +154,105 @@ func TestFwdMgr_AccountUnauthorizedToForward_SkipsForwarding(t *testing.T) {
 	err = fwdMgr.Close()
 	require.NoError(t, err)
 }
+
+func TestFwdMgr_InvalidForwarderForOCR2FeedsStates(t *testing.T) {
+	lggr := logger.Test(t)
+	db := pgtest.NewSqlxDB(t)
+	ctx := testutils.Context(t)
+	cfg := configtest.NewTestGeneralConfig(t)
+	evmcfg := evmtest.NewChainScopedConfig(t, cfg)
+	owner := testutils.MustNewSimTransactor(t)
+	ec := backends.NewSimulatedBackend(map[common.Address]core.GenesisAccount{
+		owner.From: {
+			Balance: big.NewInt(0).Mul(big.NewInt(10), big.NewInt(1e18)),
+		},
+	}, 10e6)
+	t.Cleanup(func() { ec.Close() })
+	linkAddr := common.HexToAddress("0x01BE23585060835E02B77ef475b0Cc51aA1e0709")
+	operatorAddr, _, _, err := operator_wrapper.DeployOperator(owner, ec, linkAddr, owner.From)
+	require.NoError(t, err)
+
+	forwarderAddr, _, forwarder, err := authorized_forwarder.DeployAuthorizedForwarder(owner, ec, linkAddr, owner.From, operatorAddr, []byte{})
+	require.NoError(t, err)
+	ec.Commit()
+
+	accessAddress, _, _, err := testocr2aggregator.DeploySimpleWriteAccessController(owner, ec)
+	require.NoError(t, err, "failed to deploy test access controller contract")
+	ocr2Address, _, ocr2, err := testocr2aggregator.DeployOCR2Aggregator(
+		owner,
+		ec,
+		linkAddr,
+		big.NewInt(0),
+		big.NewInt(10),
+		accessAddress,
+		accessAddress,
+		9,
+		"TEST",
+	)
+	require.NoError(t, err, "failed to deploy ocr2 test aggregator")
+	ec.Commit()
+
+	evmClient := client.NewSimulatedBackendClient(t, ec, testutils.FixtureChainID)
+	lpOpts := logpoller.Opts{
+		PollPeriod:               100 * time.Millisecond,
+		FinalityDepth:            2,
+		BackfillBatchSize:        3,
+		RpcBatchSize:             2,
+		KeepFinalizedBlocksDepth: 1000,
+	}
+	lp := logpoller.NewLogPoller(logpoller.NewORM(testutils.FixtureChainID, db, lggr), evmClient, lggr, lpOpts)
+	fwdMgr := forwarders.NewFwdMgr(db, evmClient, lp, lggr, evmcfg.EVM())
+	fwdMgr.ORM = forwarders.NewORM(db)
+
+	_, err = fwdMgr.ORM.CreateForwarder(ctx, forwarderAddr, ubig.Big(*testutils.FixtureChainID))
+	require.NoError(t, err)
+	lst, err := fwdMgr.ORM.FindForwardersByChain(ctx, ubig.Big(*testutils.FixtureChainID))
+	require.NoError(t, err)
+	require.Equal(t, len(lst), 1)
+	require.Equal(t, lst[0].Address, forwarderAddr)
+
+	fwdMgr = forwarders.NewFwdMgr(db, evmClient, lp, lggr, evmcfg.EVM())
+	require.NoError(t, fwdMgr.Start(testutils.Context(t)))
+	// cannot find forwarder because it isn't authorized nor added as a transmitter
+	addr, err := fwdMgr.ForwarderForOCR2Feeds(owner.From, ocr2Address)
+	require.ErrorContains(t, err, "Cannot find forwarder for given EOA")
+	require.True(t, utils.IsZero(addr))
+
+	_, err = forwarder.SetAuthorizedSenders(owner, []common.Address{owner.From})
+	require.NoError(t, err)
+	ec.Commit()
+
+	authorizedSenders, err := forwarder.GetAuthorizedSenders(&bind.CallOpts{Context: ctx})
+	require.NoError(t, err)
+	require.Equal(t, owner.From, authorizedSenders[0])
+
+	// cannot find forwarder because it isn't added as a transmitter
+	addr, err = fwdMgr.ForwarderForOCR2Feeds(owner.From, ocr2Address)
+	require.ErrorContains(t, err, "Cannot find forwarder for given EOA")
+	require.True(t, utils.IsZero(addr))
+
+	onchainConfig, err := testhelpers.GenerateDefaultOCR2OnchainConfig(big.NewInt(0), big.NewInt(10))
+	require.NoError(t, err)
+
+	_, err = ocr2.SetConfig(owner,
+		[]common.Address{testutils.NewAddress(), testutils.NewAddress(), testutils.NewAddress(), testutils.NewAddress()},
+		[]common.Address{forwarderAddr, testutils.NewAddress(), testutils.NewAddress(), testutils.NewAddress()},
+		1,
+		onchainConfig,
+		0,
+		[]byte{})
+	require.NoError(t, err)
+	ec.Commit()
+
+	transmitters, err := ocr2.GetTransmitters(&bind.CallOpts{Context: ctx})
+	require.NoError(t, err)
+	require.True(t, slices.Contains(transmitters, forwarderAddr))
+
+	// create new fwd to have an empty cache that has to fetch authorized forwarders from log poller
+	fwdMgr = forwarders.NewFwdMgr(db, evmClient, lp, lggr, evmcfg.EVM())
+	require.NoError(t, fwdMgr.Start(testutils.Context(t)))
+	addr, err = fwdMgr.ForwarderForOCR2Feeds(owner.From, ocr2Address)
+	require.NoError(t, err, "forwarder should be valid and found because it is both authorized and set as a transmitter")
+	require.Equal(t, forwarderAddr, addr)
+	require.NoError(t, fwdMgr.Close())
+}

core/scripts/go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/prometheus/client_golang v1.17.0
 	github.com/shopspring/decimal v1.3.1
 	github.com/smartcontractkit/chainlink-automation v1.0.3
-	github.com/smartcontractkit/chainlink-common v0.1.7-0.20240514153505-0ddba5aa4d2c
+	github.com/smartcontractkit/chainlink-common v0.1.7-0.20240516150131-e1be553a9d10
 	github.com/smartcontractkit/chainlink-vrf v0.0.0-20240222010609-cd67d123c772
 	github.com/smartcontractkit/chainlink/v2 v2.0.0-00010101000000-000000000000
 	github.com/smartcontractkit/libocr v0.0.0-20240419185742-fd3cab206b2c

core/scripts/go.sum

@@ -1185,8 +1185,8 @@ github.com/smartcontractkit/chain-selectors v1.0.10 h1:t9kJeE6B6G+hKD0GYR4kGJSCq
 github.com/smartcontractkit/chain-selectors v1.0.10/go.mod h1:d4Hi+E1zqjy9HqMkjBE5q1vcG9VGgxf5VxiRHfzi2kE=
 github.com/smartcontractkit/chainlink-automation v1.0.3 h1:h/ijT0NiyV06VxYVgcNfsE3+8OEzT3Q0Z9au0z1BPWs=
 github.com/smartcontractkit/chainlink-automation v1.0.3/go.mod h1:RjboV0Qd7YP+To+OrzHGXaxUxoSONveCoAK2TQ1INLU=
-github.com/smartcontractkit/chainlink-common v0.1.7-0.20240514153505-0ddba5aa4d2c h1:KiG8PAwUrdYn/AGBQ+B4p6erEUbEB+g6LJKhAaDjJ2s=
-github.com/smartcontractkit/chainlink-common v0.1.7-0.20240514153505-0ddba5aa4d2c/go.mod h1:sj0pjL+METqeYL9ibp0T8SXquymlaQsofa6bdfLgXX8=
+github.com/smartcontractkit/chainlink-common v0.1.7-0.20240516150131-e1be553a9d10 h1:IwJKWZHPBJbbh4oI3BGX8VNT3c/ChNiPZ/XI4iq6c0E=
+github.com/smartcontractkit/chainlink-common v0.1.7-0.20240516150131-e1be553a9d10/go.mod h1:sj0pjL+METqeYL9ibp0T8SXquymlaQsofa6bdfLgXX8=
 github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240508101745-af1ed7bc8a69 h1:Sec/GpBpUVaTEax1kSHlTvkzF/+d3w5roAQXaj5+SLA=
 github.com/smartcontractkit/chainlink-cosmos v0.4.1-0.20240508101745-af1ed7bc8a69/go.mod h1:ZQKf+0OLzCLYIisH/OdOIQuFRI6bDuw+jPBTATyHfFM=
 github.com/smartcontractkit/chainlink-data-streams v0.0.0-20240220203239-09be0ea34540 h1:xFSv8561jsLtF6gYZr/zW2z5qUUAkcFkApin2mnbYTo=

core/services/ocr2/delegate.go

@@ -496,14 +496,22 @@ func GetEVMEffectiveTransmitterID(jb *job.Job, chain legacyevm.Chain, lggr logge
 		if chain == nil {
 			return "", fmt.Errorf("job forwarding requires non-nil chain")
 		}
-		effectiveTransmitterID, err := chain.TxManager().GetForwarderForEOA(common.HexToAddress(spec.TransmitterID.String))
+
+		var err error
+		var effectiveTransmitterID common.Address
+		// Median forwarders need special handling because of OCR2Aggregator transmitters whitelist.
+		if spec.PluginType == types.Median {
+			effectiveTransmitterID, err = chain.TxManager().GetForwarderForEOAOCR2Feeds(common.HexToAddress(spec.TransmitterID.String), common.HexToAddress(spec.ContractID))
+		} else {
+			effectiveTransmitterID, err = chain.TxManager().GetForwarderForEOA(common.HexToAddress(spec.TransmitterID.String))
+		}
+
 		if err == nil {
 			return effectiveTransmitterID.String(), nil
 		} else if !spec.TransmitterID.Valid {
 			return "", errors.New("failed to get forwarder address and transmitterID is not set")
 		}
 		lggr.Warnw("Skipping forwarding for job, will fallback to default behavior", "job", jb.Name, "err", err)
-		// this shouldn't happen unless behaviour above was changed
 	}
 
 	return spec.TransmitterID.String, nil