feat: validate tx onchain

assembler/bsc_assembler.go

@@ -4,9 +4,10 @@ import (
 	"bytes"
 	"encoding/hex"
 	"fmt"
+	"time"
+
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	oracletypes "github.com/cosmos/cosmos-sdk/x/oracle/types"
-	"time"
 
 	"github.com/bnb-chain/greenfield-relayer/common"
 	"github.com/bnb-chain/greenfield-relayer/config"

assembler/greenfield_assembler.go

@@ -4,10 +4,11 @@ import (
 	"bytes"
 	"encoding/hex"
 	"fmt"
-	sdk "github.com/cosmos/cosmos-sdk/types"
 	"sync"
 	"time"
 
+	sdk "github.com/cosmos/cosmos-sdk/types"
+
 	"github.com/bnb-chain/greenfield-relayer/common"
 	"github.com/bnb-chain/greenfield-relayer/config"
 	"github.com/bnb-chain/greenfield-relayer/db"

db/dao/bsc_dao.go

@@ -2,9 +2,9 @@ package dao
 
 import (
 	"database/sql"
-	"github.com/cometbft/cometbft/votepool"
 	"time"
 
+	"github.com/cometbft/cometbft/votepool"
 	"gorm.io/gorm"
 
 	"github.com/bnb-chain/greenfield-relayer/db"

executor/greenfield_executor.go

@@ -386,3 +386,17 @@ func (e *GreenfieldExecutor) getGasLimitAndFeeAmount(msg *oracletypes.MsgClaim)
 	}
 	return e.config.GreenfieldConfig.GasLimit, e.config.GreenfieldConfig.FeeAmount, nil
 }
+
+func (e *GreenfieldExecutor) GetCrossTxPack(destChainID sdk.ChainID, channelID types.ChannelId, sequence uint64) (pack []byte, err error) {
+	return pack, retry.Do(func() error {
+		ctx, cancel := context.WithTimeout(context.Background(), RPCTimeout)
+		defer cancel()
+		pack, err = e.GetGnfdClient().GetCrossChainPackage(ctx, destChainID, uint32(channelID), sequence)
+		return err
+	}, relayercommon.RtyAttem,
+		relayercommon.RtyDelay,
+		relayercommon.RtyErr,
+		retry.OnRetry(func(n uint, err error) {
+			logging.Logger.Errorf("failed to query crosschain tx for channel %d, seq %d, attempt: %d times, max_attempts: %d", channelID, n+1, relayercommon.RtyAttNum)
+		}))
+}

listener/bsc_listener.go

@@ -7,9 +7,8 @@ import (
 	"strings"
 	"time"
 
-	sdk "github.com/cosmos/cosmos-sdk/types"
-
 	"github.com/cometbft/cometbft/votepool"
+	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	ethcommon "github.com/ethereum/go-ethereum/common"

listener/event_parser.go

@@ -3,9 +3,9 @@ package listener
 import (
 	"encoding/hex"
 	"fmt"
-	sdk "github.com/cosmos/cosmos-sdk/types"
 	"math/big"
 
+	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/core/types"
 

listener/greenfield_listener.go

@@ -8,10 +8,12 @@ import (
 	"sync"
 	"time"
 
+	"github.com/bnb-chain/greenfield-relayer/types"
 	abci "github.com/cometbft/cometbft/abci/types"
 	ctypes "github.com/cometbft/cometbft/rpc/core/types"
 	tmtypes "github.com/cometbft/cometbft/types"
 	"github.com/cometbft/cometbft/votepool"
+	sdk "github.com/cosmos/cosmos-sdk/types"
 
 	"github.com/bnb-chain/greenfield-relayer/common"
 	"github.com/bnb-chain/greenfield-relayer/config"
@@ -84,11 +86,23 @@ func (l *GreenfieldListener) poll() error {
 		case tx := <-relayTxCh:
 			txs = append(txs, tx)
 		case <-waitCh:
+			// validate tx against Chain
+			for _, tx := range txs {
+				onchainPack, err := l.greenfieldExecutor.GetCrossTxPack(sdk.ChainID(tx.DestChainId), types.ChannelId(tx.ChannelId), tx.Sequence)
+				if err != nil {
+					return fmt.Errorf("failed to get on chain tx, err=%s", err.Error())
+				}
+				if err := l.validateTx(onchainPack, tx); err != nil {
+					return fmt.Errorf("failed to validate tx, err=%s", err.Error())
+				}
+			}
+
 			b := &model.GreenfieldBlock{
 				Chain:     block.ChainID,
 				Height:    uint64(block.Height),
 				BlockTime: block.Time.Unix(),
 			}
+
 			if err := l.DaoManager.GreenfieldDao.SaveBlockAndBatchTransactions(b, txs); err != nil {
 				return fmt.Errorf("failed to persist block and tx to DB, err=%s", err.Error())
 			}
@@ -98,6 +112,40 @@ func (l *GreenfieldListener) poll() error {
 	}
 }
 
+func (l *GreenfieldListener) validateTx(expectedPack []byte, tx *model.GreenfieldRelayTransaction) error {
+	relayerFee, err := util.StrToBigInt(tx.RelayerFee)
+	if err != nil {
+		return err
+	}
+	ackRelayerFee, err := util.StrToBigInt(tx.AckRelayerFee)
+	if err != nil {
+		return err
+	}
+	// package
+	packBz := make([]byte, 0)
+
+	// package header
+	packageHeader := sdk.EncodePackageHeader(sdk.PackageHeader{
+		PackageType:   sdk.CrossChainPackageType(tx.PackageType),
+		Timestamp:     uint64(tx.TxTime),
+		RelayerFee:    relayerFee,
+		AckRelayerFee: ackRelayerFee,
+	})
+	packBz = append(packBz, packageHeader...)
+
+	// package payload
+	payloadBz, err := hex.DecodeString(tx.PayLoad)
+	if err != nil {
+		return err
+	}
+	packBz = append(packBz, payloadBz...)
+
+	if !bytes.Equal(expectedPack, packBz) {
+		return fmt.Errorf("package not match")
+	}
+	return nil
+}
+
 func (l *GreenfieldListener) getLatestPolledBlock() (*model.GreenfieldBlock, error) {
 	return l.DaoManager.GreenfieldDao.GetLatestBlock()
 }

main.go

@@ -3,12 +3,14 @@ package main
 import (
 	"flag"
 	"fmt"
+	"os"
+
+	"github.com/spf13/pflag"
+	"github.com/spf13/viper"
+
 	"github.com/bnb-chain/greenfield-relayer/app"
 	"github.com/bnb-chain/greenfield-relayer/config"
 	"github.com/bnb-chain/greenfield-relayer/logging"
-	"github.com/spf13/pflag"
-	"github.com/spf13/viper"
-	"os"
 )
 
 func initFlags() {

util/util.go

@@ -2,6 +2,7 @@ package util
 
 import (
 	"encoding/binary"
+	"fmt"
 	"math/big"
 	"strconv"
 
@@ -41,3 +42,12 @@ func Uint64ToBytes(num uint64) []byte {
 	binary.BigEndian.PutUint64(bt, num)
 	return bt
 }
+
+func StrToBigInt(str string) (*big.Int, error) {
+	var r big.Int
+	num, ok := r.SetString(str, 10)
+	if !ok {
+		return &big.Int{}, fmt.Errorf("convetion failed, %s", str)
+	}
+	return num, nil
+}