feat: custom pubkeys inmemory service

aggregator/aggregator.go

@@ -8,17 +8,15 @@ import (
  "sync"
  "time"
 
- chainioavsregistry "github.com/Layr-Labs/eigensdk-go/chainio/clients/avsregistry"
  "github.com/Layr-Labs/eigensdk-go/chainio/clients/wallet"
  "github.com/Layr-Labs/eigensdk-go/chainio/txmgr"
+ "github.com/Layr-Labs/eigensdk-go/crypto/bls"
  "github.com/Layr-Labs/eigensdk-go/logging"
  "github.com/Layr-Labs/eigensdk-go/metrics"
  "github.com/Layr-Labs/eigensdk-go/services/avsregistry"
  blsagg "github.com/Layr-Labs/eigensdk-go/services/bls_aggregation"
- opinfoserv "github.com/Layr-Labs/eigensdk-go/services/operatorsinfo"
  "github.com/Layr-Labs/eigensdk-go/signerv2"
  eigentypes "github.com/Layr-Labs/eigensdk-go/types"
- "github.com/ethereum/go-ethereum/common"
  "github.com/prometheus/client_golang/prometheus"
 
  "github.com/NethermindEth/near-sffl/aggregator/database"
@@ -46,6 +44,9 @@ var (
  DigestError = errors.New("Failed to get message digest")
  TaskResponseDigestError = errors.New("Failed to get task response digest")
  GetOperatorSetUpdateBlockError = errors.New("Failed to get operator set update block")
+ OperatorNotFoundError = errors.New("Operator not found")
+ InvalidSignatureError = errors.New("Invalid signature")
+ UnsupportedMessageTypeError = errors.New("Unsupported message type")
 
  // REST errors
  StateRootUpdateNotFoundError = errors.New("StateRootUpdate not found")
@@ -61,6 +62,7 @@ type RpcAggregatorer interface {
  ProcessSignedOperatorSetUpdateMessage(signedOperatorSetUpdateMessage *messages.SignedOperatorSetUpdateMessage) error
  GetAggregatedCheckpointMessages(fromTimestamp, toTimestamp uint64) (*messages.CheckpointMessages, error)
  GetRegistryCoordinatorAddress(reply *string) error
+ GetOperatorInfoById(ctx context.Context, operatorId eigentypes.OperatorId) (eigentypes.OperatorInfo, bool)
 }
 
 type RestAggregatorer interface {
@@ -120,6 +122,7 @@ type Aggregator struct {
  metrics metrics.Metrics
  aggregatorListener AggregatorEventListener
 
+ operatorRegistrationsService OperatorRegistrationsService
  taskBlsAggregationService blsagg.BlsAggregationService
  stateRootUpdateBlsAggregationService MessageBlsAggregationService
  operatorSetUpdateBlsAggregationService MessageBlsAggregationService
@@ -184,13 +187,6 @@ func NewAggregator(
 
  txMgr := txmgr.NewSimpleTxManager(txSender, ethHttpClient, logger, config.AggregatorAddress).WithGasLimitMultiplier(1.5)
 
- // note that the subscriber needs a ws connection instead of http
- avsRegistryChainSubscriber, err := chainioavsregistry.BuildAvsRegistryChainSubscriber(common.HexToAddress(config.SFFLRegistryCoordinatorAddr.String()), ethWsClient, logger)
- if err != nil {
- logger.Error("Cannot create AvsRegistryChainSubscriber", "err", err)
- return nil, err
- }
-
  avsWriter, err := chainio.BuildAvsWriterFromConfig(txMgr, config, ethHttpClient, logger)
  if err != nil {
  logger.Error("Cannot create avsWriter", "err", err)
@@ -215,8 +211,12 @@ func NewAggregator(
  return nil, err
  }
 
- operatorPubkeysService := opinfoserv.NewOperatorsInfoServiceInMemory(ctx, avsRegistryChainSubscriber, avsReader, logger)
- avsRegistryService := avsregistry.NewAvsRegistryServiceChainCaller(avsReader, operatorPubkeysService, logger)
+ operatorRegistrationsService, err := NewOperatorRegistrationsServiceInMemory(ctx, avsSubscriber, avsReader, logger)
+ if err != nil {
+ return nil, err
+ }
+
+ avsRegistryService := avsregistry.NewAvsRegistryServiceChainCaller(avsReader, operatorRegistrationsService, logger)
  taskBlsAggregationService := blsagg.NewBlsAggregatorService(avsRegistryService, logger)
  stateRootUpdateBlsAggregationService := NewMessageBlsAggregatorService(avsRegistryService, ethHttpClient, logger)
  operatorSetUpdateBlsAggregationService := NewMessageBlsAggregatorService(avsRegistryService, ethHttpClient, logger)
@@ -232,6 +232,7 @@ func NewAggregator(
  rollupBroadcaster: rollupBroadcaster,
  httpClient: ethHttpClient,
  wsClient: ethWsClient,
+ operatorRegistrationsService: operatorRegistrationsService,
  clock: core.SystemClock,
  taskBlsAggregationService: taskBlsAggregationService,
  stateRootUpdateBlsAggregationService: stateRootUpdateBlsAggregationService,
@@ -522,6 +523,11 @@ func (agg *Aggregator) handleOperatorSetUpdateReachedQuorum(ctx context.Context,
 }
 
 func (agg *Aggregator) ProcessSignedCheckpointTaskResponse(signedCheckpointTaskResponse *messages.SignedCheckpointTaskResponse) error {
+ err := agg.verifySignature(signedCheckpointTaskResponse)
+ if err != nil {
+ return err
+ }
+
  taskIndex := signedCheckpointTaskResponse.TaskResponse.ReferenceTaskIndex
  taskResponseDigest, err := signedCheckpointTaskResponse.TaskResponse.Digest()
  if err != nil {
@@ -551,19 +557,24 @@ func (agg *Aggregator) ProcessSignedCheckpointTaskResponse(signedCheckpointTaskR
 
 // Rpc request handlers
 func (agg *Aggregator) ProcessSignedStateRootUpdateMessage(signedStateRootUpdateMessage *messages.SignedStateRootUpdateMessage) error {
- messageDigest, err := signedStateRootUpdateMessage.Message.Digest()
+ timestamp := signedStateRootUpdateMessage.Message.Timestamp
+ err := agg.validateMessageTimestamp(timestamp)
  if err != nil {
- agg.logger.Error("Failed to get message digest", "err", err)
- return DigestError
+ agg.logger.Error("Failed to validate message timestamp", "err", err, "timestamp", timestamp)
+ return err
  }
 
- timestamp := signedStateRootUpdateMessage.Message.Timestamp
- err = agg.validateMessageTimestamp(timestamp)
+ err = agg.verifySignature(signedStateRootUpdateMessage)
  if err != nil {
- agg.logger.Error("Failed to validate message timestamp", "err", err, "timestamp", timestamp)
  return err
  }
 
+ messageDigest, err := signedStateRootUpdateMessage.Message.Digest()
+ if err != nil {
+ agg.logger.Error("Failed to get message digest", "err", err)
+ return DigestError
+ }
+
  err = agg.stateRootUpdateBlsAggregationService.InitializeMessageIfNotExists(
  messageDigest,
  coretypes.QUORUM_NUMBERS,
@@ -588,19 +599,24 @@ func (agg *Aggregator) ProcessSignedStateRootUpdateMessage(signedStateRootUpdate
 }
 
 func (agg *Aggregator) ProcessSignedOperatorSetUpdateMessage(signedOperatorSetUpdateMessage *messages.SignedOperatorSetUpdateMessage) error {
- messageDigest, err := signedOperatorSetUpdateMessage.Message.Digest()
+ timestamp := signedOperatorSetUpdateMessage.Message.Timestamp
+ err := agg.validateMessageTimestamp(timestamp)
  if err != nil {
- agg.logger.Error("Failed to get message digest", "err", err)
- return DigestError
+ agg.logger.Error("Failed to validate message timestamp", "err", err, "timestamp", timestamp)
+ return err
  }
 
- timestamp := signedOperatorSetUpdateMessage.Message.Timestamp
- err = agg.validateMessageTimestamp(timestamp)
+ err = agg.verifySignature(signedOperatorSetUpdateMessage)
  if err != nil {
- agg.logger.Error("Failed to validate message timestamp", "err", err, "timestamp", timestamp)
  return err
  }
 
+ messageDigest, err := signedOperatorSetUpdateMessage.Message.Digest()
+ if err != nil {
+ agg.logger.Error("Failed to get message digest", "err", err)
+ return DigestError
+ }
+
  blockNumber, err := agg.avsReader.GetOperatorSetUpdateBlock(context.Background(), signedOperatorSetUpdateMessage.Message.Id)
  if err != nil {
  agg.logger.Error("Failed to get operator set update block", "err", err)
@@ -691,6 +707,59 @@ func (agg *Aggregator) GetCheckpointMessages(fromTimestamp, toTimestamp uint64)
  }, nil
 }
 
+func (agg *Aggregator) GetOperatorInfoById(ctx context.Context, operatorId eigentypes.OperatorId) (eigentypes.OperatorInfo, bool) {
+ operatorInfo, ok := agg.operatorRegistrationsService.GetOperatorInfoById(ctx, operatorId)
+ return operatorInfo, ok
+}
+
+func (agg *Aggregator) verifySignature(signedMessage interface{}) error {
+ var operatorId eigentypes.OperatorId
+ var signature bls.Signature
+ var digest [32]byte
+ var err error
+
+ switch signedMessage := signedMessage.(type) {
+ case *messages.SignedCheckpointTaskResponse:
+ operatorId = signedMessage.OperatorId
+ signature = signedMessage.BlsSignature
+ digest, err = signedMessage.TaskResponse.Digest()
+ if err != nil {
+ return TaskResponseDigestError
+ }
+ case *messages.SignedStateRootUpdateMessage:
+ operatorId = signedMessage.OperatorId
+ signature = signedMessage.BlsSignature
+ digest, err = signedMessage.Message.Digest()
+ if err != nil {
+ return DigestError
+ }
+ case *messages.SignedOperatorSetUpdateMessage:
+ operatorId = signedMessage.OperatorId
+ signature = signedMessage.BlsSignature
+ digest, err = signedMessage.Message.Digest()
+ if err != nil {
+ return DigestError
+ }
+ default:
+ return UnsupportedMessageTypeError
+ }
+
+ operatorInfo, ok := agg.GetOperatorInfoById(context.Background(), operatorId)
+ if !ok {
+ return OperatorNotFoundError
+ }
+
+ ok, err = signature.Verify(operatorInfo.Pubkeys.G2Pubkey, digest)
+ if err != nil {
+ return InvalidSignatureError
+ }
+ if !ok {
+ return InvalidSignatureError
+ }
+
+ return nil
+}
+
 func (agg *Aggregator) validateMessageTimestamp(messageTimestamp uint64) error {
  now := agg.clock.Now().Unix()
  timeoutInSeconds := types.MESSAGE_SUBMISSION_TIMEOUT.Seconds()

aggregator/aggregator_test.go

@@ -34,13 +34,14 @@ var MOCK_OPERATOR_BLS_PRIVATE_KEY, _ = bls.NewPrivateKey(MOCK_OPERATOR_BLS_PRIVA
 var MOCK_OPERATOR_KEYPAIR = bls.NewKeyPair(MOCK_OPERATOR_BLS_PRIVATE_KEY)
 var MOCK_OPERATOR_G1PUBKEY = MOCK_OPERATOR_KEYPAIR.GetPubKeyG1()
 var MOCK_OPERATOR_G2PUBKEY = MOCK_OPERATOR_KEYPAIR.GetPubKeyG2()
+var MOCK_OPERATOR_PUBKEYS = eigentypes.OperatorPubkeys{
+ G1Pubkey: MOCK_OPERATOR_G1PUBKEY,
+ G2Pubkey: MOCK_OPERATOR_G2PUBKEY,
+}
 var MOCK_OPERATOR_PUBKEY_DICT = map[eigentypes.OperatorId]types.OperatorInfo{
  MOCK_OPERATOR_ID: {
- OperatorPubkeys: eigentypes.OperatorPubkeys{
- G1Pubkey: MOCK_OPERATOR_G1PUBKEY,
- G2Pubkey: MOCK_OPERATOR_G2PUBKEY,
- },
- OperatorAddr: common.Address{},
+ OperatorPubkeys: MOCK_OPERATOR_PUBKEYS,
+ OperatorAddr: common.Address{},
  },
 }
 
@@ -55,7 +56,7 @@ func TestSendNewTask(t *testing.T) {
  mockCtrl := gomock.NewController(t)
  defer mockCtrl.Finish()
 
- aggregator, mockAvsReaderer, mockAvsWriterer, mockTaskBlsAggService, _, _, _, _, mockClient, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
+ aggregator, mockAvsReaderer, mockAvsWriterer, mockTaskBlsAggService, _, _, _, _, _, mockClient, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
  assert.Nil(t, err)
 
  var TASK_INDEX = uint32(0)
@@ -92,7 +93,7 @@ func TestHandleStateRootUpdateAggregationReachedQuorum(t *testing.T) {
  mockCtrl := gomock.NewController(t)
  defer mockCtrl.Finish()
 
- aggregator, _, _, _, _, _, mockMsgDb, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
+ aggregator, _, _, _, _, _, _, mockMsgDb, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
  assert.Nil(t, err)
 
  msg := messages.StateRootUpdateMessage{}
@@ -122,7 +123,7 @@ func TestHandleOperatorSetUpdateAggregationReachedQuorum(t *testing.T) {
  mockCtrl := gomock.NewController(t)
  defer mockCtrl.Finish()
 
- aggregator, _, _, _, _, _, mockMsgDb, mockRollupBroadcaster, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
+ aggregator, _, _, _, _, _, _, mockMsgDb, mockRollupBroadcaster, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
  assert.Nil(t, err)
 
  msg := messages.OperatorSetUpdateMessage{}
@@ -158,7 +159,7 @@ func TestExpiredStateRootUpdateMessage(t *testing.T) {
  mockCtrl := gomock.NewController(t)
  defer mockCtrl.Finish()
 
- aggregator, _, _, _, _, _, _, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
+ aggregator, _, _, _, _, _, _, _, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
  assert.NoError(t, err)
 
  nowTimestamp := uint64(6000)
@@ -178,7 +179,7 @@ func TestExpiredOperatorSetUpdate(t *testing.T) {
  mockCtrl := gomock.NewController(t)
  defer mockCtrl.Finish()
 
- aggregator, _, _, _, _, _, _, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
+ aggregator, _, _, _, _, _, _, _, _, _, err := createMockAggregator(mockCtrl, MOCK_OPERATOR_PUBKEY_DICT)
  assert.NoError(t, err)
 
  nowTimestamp := uint64(8000)
@@ -196,7 +197,7 @@ func TestExpiredOperatorSetUpdate(t *testing.T) {
 
 func createMockAggregator(
  mockCtrl *gomock.Controller, operatorPubkeyDict map[eigentypes.OperatorId]types.OperatorInfo,
-) (*Aggregator, *chainiomocks.MockAvsReaderer, *chainiomocks.MockAvsWriterer, *blsaggservmock.MockBlsAggregationService, *aggmocks.MockMessageBlsAggregationService, *aggmocks.MockMessageBlsAggregationService, *dbmocks.MockDatabaser, *aggmocks.MockRollupBroadcasterer, *safeclientmocks.MockSafeClient, error) {
+) (*Aggregator, *chainiomocks.MockAvsReaderer, *chainiomocks.MockAvsWriterer, *blsaggservmock.MockBlsAggregationService, *aggmocks.MockMessageBlsAggregationService, *aggmocks.MockMessageBlsAggregationService, *aggmocks.MockOperatorRegistrationsService, *dbmocks.MockDatabaser, *aggmocks.MockRollupBroadcasterer, *safeclientmocks.MockSafeClient, error) {
  logger := sdklogging.NewNoopLogger()
  mockAvsWriter := chainiomocks.NewMockAvsWriterer(mockCtrl)
  mockAvsReader := chainiomocks.NewMockAvsReaderer(mockCtrl)
@@ -206,6 +207,7 @@ func createMockAggregator(
  mockMsgDb := dbmocks.NewMockDatabaser(mockCtrl)
  mockRollupBroadcaster := aggmocks.NewMockRollupBroadcasterer(mockCtrl)
  mockClient := safeclientmocks.NewMockSafeClient(mockCtrl)
+ mockOperatorRegistrationsService := aggmocks.NewMockOperatorRegistrationsService(mockCtrl)
 
  aggregator := &Aggregator{
  logger: logger,
@@ -214,6 +216,7 @@ func createMockAggregator(
  taskBlsAggregationService: mockTaskBlsAggregationService,
  stateRootUpdateBlsAggregationService: mockStateRootUpdateBlsAggregationService,
  operatorSetUpdateBlsAggregationService: mockOperatorSetUpdateBlsAggregationService,
+ operatorRegistrationsService: mockOperatorRegistrationsService,
  msgDb: mockMsgDb,
  tasks: make(map[coretypes.TaskIndex]taskmanager.CheckpointTask),
  taskResponses: make(map[coretypes.TaskIndex]map[eigentypes.TaskResponseDigest]messages.CheckpointTaskResponse),
@@ -225,5 +228,5 @@ func createMockAggregator(
  aggregatorListener: &SelectiveAggregatorListener{},
  clock: core.SystemClock,
  }
- return aggregator, mockAvsReader, mockAvsWriter, mockTaskBlsAggregationService, mockStateRootUpdateBlsAggregationService, mockOperatorSetUpdateBlsAggregationService, mockMsgDb, mockRollupBroadcaster, mockClient, nil
+ return aggregator, mockAvsReader, mockAvsWriter, mockTaskBlsAggregationService, mockStateRootUpdateBlsAggregationService, mockOperatorSetUpdateBlsAggregationService, mockOperatorRegistrationsService, mockMsgDb, mockRollupBroadcaster, mockClient, nil
 }

aggregator/gen.go

@@ -4,4 +4,5 @@ package aggregator
 //go:generate mockgen -destination=./mocks/rpc_aggregator.go -package=mocks github.com/NethermindEth/near-sffl/aggregator RpcAggregatorer
 //go:generate mockgen -destination=./mocks/message_blsagg.go -package=mocks github.com/NethermindEth/near-sffl/aggregator MessageBlsAggregationService
 //go:generate mockgen -destination=./mocks/rollup_broadcaster.go -package=mocks github.com/NethermindEth/near-sffl/aggregator RollupBroadcasterer
+//go:generate mockgen -destination=./mocks/operator_registrations_inmemory.go -package=mocks github.com/NethermindEth/near-sffl/aggregator OperatorRegistrationsService
 //go:generate mockgen -destination=./mocks/eth_client.go -package=mocks github.com/Layr-Labs/eigensdk-go/chainio/clients/eth Client