Integrate fireblocks wallet into ejector (#527)

core/eth/tx.go

@@ -364,7 +364,7 @@ func (t *Transactor) UpdateOperatorSocket(ctx context.Context, socket string) er
 	return nil
 }
 
-func (t *Transactor) EjectOperators(ctx context.Context, operatorsByQuorum [][]core.OperatorID) (*types.Receipt, error) {
+func (t *Transactor) BuildEjectOperatorsTxn(ctx context.Context, operatorsByQuorum [][]core.OperatorID) (*types.Transaction, error) {
 	byteIdsByQuorum := make([][][32]byte, len(operatorsByQuorum))
 	for i, ids := range operatorsByQuorum {
 		for _, id := range ids {
@@ -376,17 +376,7 @@ func (t *Transactor) EjectOperators(ctx context.Context, operatorsByQuorum [][]c
 		t.Logger.Error("Failed to generate transact opts", "err", err)
 		return nil, err
 	}
-	txn, err := t.Bindings.EjectionManager.EjectOperators(opts, byteIdsByQuorum)
-	if err != nil {
-		t.Logger.Error("Failed to create transaction", "err", err)
-		return nil, err
-	}
-	receipt, err := t.EthClient.EstimateGasPriceAndLimitAndSendTx(context.Background(), txn, "EjectOperators", nil)
-	if err != nil {
-		t.Logger.Error("Failed to eject operators", "err", err)
-		return nil, err
-	}
-	return receipt, nil
+	return t.Bindings.EjectionManager.EjectOperators(opts, byteIdsByQuorum)
 }
 
 // GetOperatorStakes returns the stakes of all operators within the quorums that the operator represented by operatorId

core/mock/tx.go

@@ -70,10 +70,10 @@ func (t *MockTransactor) UpdateOperatorSocket(ctx context.Context, socket string
 	return args.Error(0)
 }
 
-func (t *MockTransactor) EjectOperators(ctx context.Context, operatorsByQuorum [][]core.OperatorID) (*types.Receipt, error) {
+func (t *MockTransactor) BuildEjectOperatorsTxn(ctx context.Context, operatorsByQuorum [][]core.OperatorID) (*types.Transaction, error) {
 	args := t.Called()
 	result := args.Get(0)
-	return result.(*types.Receipt), args.Error(1)
+	return result.(*types.Transaction), args.Error(1)
 }
 
 func (t *MockTransactor) GetOperatorStakes(ctx context.Context, operatorId core.OperatorID, blockNumber uint32) (core.OperatorStakes, []core.QuorumID, error) {

core/tx.go

@@ -70,10 +70,10 @@ type Transactor interface {
 	// UpdateOperatorSocket updates the socket of the operator in all the quorums that it is registered with.
 	UpdateOperatorSocket(ctx context.Context, socket string) error
 
-	// EjectOperators ejects operators from AVS registryCoordinator.
+	// BuildEjectOperatorsTxn returns a transaction that ejects operators from AVS registryCoordinator.
 	// The operatorsByQuorum provides a list of operators for each quorum. Within a quorum,
 	// the operators are ordered; in case of rate limiting, the first operators will be ejected.
-	EjectOperators(ctx context.Context, operatorsByQuorum [][]OperatorID) (*types.Receipt, error)
+	BuildEjectOperatorsTxn(ctx context.Context, operatorsByQuorum [][]OperatorID) (*types.Transaction, error)
 
 	// GetOperatorStakes returns the stakes of all operators within the quorums that the operator represented by operatorId
 	//  is registered with. The returned stakes are for the block number supplied. The indices of the operators within each quorum

disperser/cmd/dataapi/config.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"errors"
+	"time"
 
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/aws"
@@ -17,6 +18,7 @@ type Config struct {
 	AwsClientConfig  aws.ClientConfig
 	BlobstoreConfig  blobstore.Config
 	EthClientConfig  geth.EthClientConfig
+	FireblocksConfig common.FireblocksConfig
 	LoggerConfig     common.LoggerConfig
 	PrometheusConfig prometheus.Config
 	MetricsConfig    dataapi.MetricsConfig
@@ -35,6 +37,9 @@ type Config struct {
 	DisperserHostname  string
 	ChurnerHostname    string
 	BatcherHealthEndpt string
+
+	FireblockAPITimeout time.Duration
+	TxnTimeout          time.Duration
 }
 
 func NewConfig(ctx *cli.Context) (Config, error) {
@@ -46,13 +51,19 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 	if len(ejectionToken) < 20 {
 		return Config{}, errors.New("the ejection token length must be at least 20")
 	}
+	ethClientConfig := geth.ReadEthClientConfig(ctx)
+	fireblocksConfig := common.ReadFireblocksCLIConfig(ctx, flags.FlagPrefix)
+	if !fireblocksConfig.Disable {
+		ethClientConfig = geth.ReadEthClientConfigRPCOnly(ctx)
+	}
 	config := Config{
 		BlobstoreConfig: blobstore.Config{
 			BucketName: ctx.GlobalString(flags.S3BucketNameFlag.Name),
 			TableName:  ctx.GlobalString(flags.DynamoTableNameFlag.Name),
 		},
 		AwsClientConfig:               aws.ReadClientConfig(ctx, flags.FlagPrefix),
-		EthClientConfig:               geth.ReadEthClientConfig(ctx),
+		EthClientConfig:               ethClientConfig,
+		FireblocksConfig:              fireblocksConfig,
 		LoggerConfig:                  *loggerConfig,
 		SocketAddr:                    ctx.GlobalString(flags.SocketAddrFlag.Name),
 		SubgraphApiBatchMetadataAddr:  ctx.GlobalString(flags.SubgraphApiBatchMetadataAddrFlag.Name),
@@ -75,6 +86,9 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 		DisperserHostname:  ctx.GlobalString(flags.DisperserHostnameFlag.Name),
 		ChurnerHostname:    ctx.GlobalString(flags.ChurnerHostnameFlag.Name),
 		BatcherHealthEndpt: ctx.GlobalString(flags.BatcherHealthEndptFlag.Name),
+
+		FireblockAPITimeout: ctx.GlobalDuration(flags.FireblockAPITimeoutFlag.Name),
+		TxnTimeout:          ctx.GlobalDuration(flags.TxnTimeoutFlag.Name),
 	}
 	return config, nil
 }

disperser/cmd/dataapi/flags/flags.go

@@ -1,6 +1,8 @@
 package flags
 
 import (
+	"time"
+
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/aws"
 	"github.com/Layr-Labs/eigenda/common/geth"
@@ -137,6 +139,20 @@ var (
 		Value:    "9100",
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "METRICS_HTTP_PORT"),
 	}
+	FireblockAPITimeoutFlag = cli.DurationFlag{
+		Name:     common.PrefixFlag(FlagPrefix, "fireblocks-api-timeout"),
+		Usage:    "the timeout for the fireblocks api",
+		Required: false,
+		Value:    3 * time.Minute,
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "FIREBLOCKS_API_TIMEOUT"),
+	}
+	TxnTimeoutFlag = cli.DurationFlag{
+		Name:     common.PrefixFlag(FlagPrefix, "txn-timeout"),
+		Usage:    "the timeout for the transaction",
+		Required: false,
+		Value:    6 * time.Minute,
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "TRANSACTION_TIMEOUT"),
+	}
 )
 
 var requiredFlags = []cli.Flag{
@@ -157,6 +173,8 @@ var requiredFlags = []cli.Flag{
 	DisperserHostnameFlag,
 	ChurnerHostnameFlag,
 	BatcherHealthEndptFlag,
+	FireblockAPITimeoutFlag,
+	TxnTimeoutFlag,
 }
 
 var optionalFlags = []cli.Flag{
@@ -171,5 +189,6 @@ func init() {
 	Flags = append(requiredFlags, optionalFlags...)
 	Flags = append(Flags, common.LoggerCLIFlags(envVarPrefix, FlagPrefix)...)
 	Flags = append(Flags, geth.EthClientFlags(envVarPrefix)...)
+	Flags = append(Flags, common.FireblocksCLIFlags(envVarPrefix, FlagPrefix)...)
 	Flags = append(Flags, aws.ClientFlags(envVarPrefix, FlagPrefix)...)
 }

disperser/cmd/dataapi/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
 	"os"
@@ -11,14 +12,21 @@ import (
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/common/aws/dynamodb"
 	"github.com/Layr-Labs/eigenda/common/aws/s3"
+	"github.com/Layr-Labs/eigenda/common/aws/secretmanager"
 	"github.com/Layr-Labs/eigenda/common/geth"
 	coreeth "github.com/Layr-Labs/eigenda/core/eth"
 	"github.com/Layr-Labs/eigenda/disperser/cmd/dataapi/flags"
 	"github.com/Layr-Labs/eigenda/disperser/common/blobstore"
 	"github.com/Layr-Labs/eigenda/disperser/dataapi"
 	"github.com/Layr-Labs/eigenda/disperser/dataapi/prometheus"
 	"github.com/Layr-Labs/eigenda/disperser/dataapi/subgraph"
+	"github.com/Layr-Labs/eigensdk-go/chainio/clients/fireblocks"
+	walletsdk "github.com/Layr-Labs/eigensdk-go/chainio/clients/wallet"
+	"github.com/Layr-Labs/eigensdk-go/logging"
+	"github.com/Layr-Labs/eigensdk-go/signerv2"
+
 	gethcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/urfave/cli"
 )
 
@@ -86,6 +94,10 @@ func RunDataApi(ctx *cli.Context) error {
 		return err
 	}
 
+	wallet, err := getWallet(config, client, logger)
+	if err != nil {
+		return err
+	}
 	var (
 		promClient        = dataapi.NewPrometheusClient(promApi, config.PrometheusConfig.Cluster)
 		blobMetadataStore = blobstore.NewBlobMetadataStore(dynamoClient, logger, config.BlobstoreConfig.TableName, 0)
@@ -109,6 +121,7 @@ func RunDataApi(ctx *cli.Context) error {
 			subgraphClient,
 			tx,
 			chainState,
+			dataapi.NewEjector(wallet, client, logger, tx, metrics, config.TxnTimeout),
 			logger,
 			metrics,
 			nil,
@@ -147,3 +160,70 @@ func RunDataApi(ctx *cli.Context) error {
 
 	return err
 }
+
+func getWallet(config Config, ethClient common.EthClient, logger logging.Logger) (walletsdk.Wallet, error) {
+	var wallet walletsdk.Wallet
+	if !config.FireblocksConfig.Disable {
+		validConfigflag := len(config.FireblocksConfig.APIKeyName) > 0 &&
+			len(config.FireblocksConfig.SecretKeyName) > 0 &&
+			len(config.FireblocksConfig.BaseURL) > 0 &&
+			len(config.FireblocksConfig.VaultAccountName) > 0 &&
+			len(config.FireblocksConfig.WalletAddress) > 0 &&
+			len(config.FireblocksConfig.Region) > 0
+		if !validConfigflag {
+			return nil, errors.New("fireblocks config is either invalid or incomplete")
+		}
+		apiKey, err := secretmanager.ReadStringFromSecretManager(context.Background(), config.FireblocksConfig.APIKeyName, config.FireblocksConfig.Region)
+		if err != nil {
+			return nil, fmt.Errorf("cannot read fireblocks api key %s from secret manager: %w", config.FireblocksConfig.APIKeyName, err)
+		}
+		secretKey, err := secretmanager.ReadStringFromSecretManager(context.Background(), config.FireblocksConfig.SecretKeyName, config.FireblocksConfig.Region)
+		if err != nil {
+			return nil, fmt.Errorf("cannot read fireblocks secret key %s from secret manager: %w", config.FireblocksConfig.SecretKeyName, err)
+		}
+		fireblocksClient, err := fireblocks.NewClient(
+			apiKey,
+			[]byte(secretKey),
+			config.FireblocksConfig.BaseURL,
+			config.FireblockAPITimeout,
+			logger.With("component", "FireblocksClient"),
+		)
+		if err != nil {
+			return nil, err
+		}
+		wallet, err = walletsdk.NewFireblocksWallet(fireblocksClient, ethClient, config.FireblocksConfig.VaultAccountName, logger.With("component", "FireblocksWallet"))
+		if err != nil {
+			return nil, err
+		}
+		sender, err := wallet.SenderAddress(context.Background())
+		if err != nil {
+			return nil, err
+		}
+		if sender.Cmp(gethcommon.HexToAddress(config.FireblocksConfig.WalletAddress)) != 0 {
+			return nil, fmt.Errorf("configured wallet address %s does not match derived address %s", config.FireblocksConfig.WalletAddress, sender.Hex())
+		}
+		logger.Info("Initialized Fireblocks wallet", "vaultAccountName", config.FireblocksConfig.VaultAccountName, "address", sender.Hex())
+	} else if len(config.EthClientConfig.PrivateKeyString) > 0 {
+		privateKey, err := crypto.HexToECDSA(config.EthClientConfig.PrivateKeyString)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse private key: %w", err)
+		}
+		chainID, err := ethClient.ChainID(context.Background())
+		if err != nil {
+			return nil, fmt.Errorf("failed to get chain ID: %w", err)
+		}
+		signerV2, address, err := signerv2.SignerFromConfig(signerv2.Config{PrivateKey: privateKey}, chainID)
+		if err != nil {
+			return nil, err
+		}
+		wallet, err = walletsdk.NewPrivateKeyWallet(ethClient, signerV2, address, logger.With("component", "PrivateKeyWallet"))
+		if err != nil {
+			return nil, err
+		}
+		logger.Info("Initialized PrivateKey wallet", "address", address.Hex())
+	} else {
+		return nil, errors.New("no wallet is configured. Either Fireblocks or PrivateKey wallet should be configured")
+	}
+
+	return wallet, nil
+}