Add /eth/v2/validator/aggregate_and_proofs

CHANGELOG.md

@@ -17,6 +17,7 @@ The format is based on Keep a Changelog, and this project adheres to Semantic Ve
 - GetBeaconStateV2: add Electra case.
 - Implement [consensus-specs/3875](https://github.com/ethereum/consensus-specs/pull/3875)
 - Tests to ensure sepolia config matches the official upstream yaml
+- Added SubmitAggregateAndProofsRequestV2 endpoint.
 
 ### Changed
 

api/server/structs/endpoints_validator.go

@@ -18,6 +18,10 @@ type SubmitAggregateAndProofsRequest struct {
 	Data []*SignedAggregateAttestationAndProof `json:"data"`
 }
 
+type SubmitAggregateAndProofsRequestV2 struct {
+	Data []json.RawMessage `json:"data"`
+}
+
 type SubmitSyncCommitteeSubscriptionsRequest struct {
 	Data []*SyncCommitteeSubscription `json:"data"`
 }

beacon-chain/rpc/endpoints.go

@@ -219,6 +219,16 @@ func (s *Service) validatorEndpoints(
 			handler: server.SubmitAggregateAndProofs,
 			methods: []string{http.MethodPost},
 		},
+		{
+			template: "/eth/v2/validator/aggregate_and_proofs",
+			name:     namespace + ".SubmitAggregateAndProofsV2",
+			middleware: []middleware.Middleware{
+				middleware.ContentTypeHandler([]string{api.JsonMediaType}),
+				middleware.AcceptHeaderHandler([]string{api.JsonMediaType}),
+			},
+			handler: server.SubmitAggregateAndProofsV2,
+			methods: []string{http.MethodPost},
+		},
 		{
 			template: "/eth/v1/validator/sync_committee_contribution",
 			name:     namespace + ".ProduceSyncCommitteeContribution",

beacon-chain/rpc/endpoints_test.go

@@ -99,6 +99,7 @@ func Test_endpoints(t *testing.T) {
 		"/eth/v1/validator/attestation_data":               {http.MethodGet},
 		"/eth/v1/validator/aggregate_attestation":          {http.MethodGet},
 		"/eth/v1/validator/aggregate_and_proofs":           {http.MethodPost},
+		"/eth/v2/validator/aggregate_and_proofs":           {http.MethodPost},
 		"/eth/v1/validator/beacon_committee_subscriptions": {http.MethodPost},
 		"/eth/v1/validator/sync_committee_subscriptions":   {http.MethodPost},
 		"/eth/v1/validator/beacon_committee_selections":    {http.MethodPost},

beacon-chain/rpc/eth/validator/BUILD.bazel

@@ -85,6 +85,7 @@ go_test(
         "//encoding/bytesutil:go_default_library",
         "//network/httputil:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",
+        "//runtime/version:go_default_library",
         "//testing/assert:go_default_library",
         "//testing/mock:go_default_library",
         "//testing/require:go_default_library",

beacon-chain/rpc/eth/validator/handlers.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v5/api"
 	"github.com/prysmaticlabs/prysm/v5/api/server/structs"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/builder"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/cache"
@@ -31,6 +32,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/monitoring/tracing/trace"
 	"github.com/prysmaticlabs/prysm/v5/network/httputil"
 	ethpbalpha "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v5/runtime/version"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc/codes"
@@ -191,6 +193,94 @@ func (s *Server) SubmitAggregateAndProofs(w http.ResponseWriter, r *http.Request
 	}
 }
 
+// SubmitAggregateAndProofsV2 verifies given aggregate and proofs and publishes them on appropriate gossipsub topic.
+func (s *Server) SubmitAggregateAndProofsV2(w http.ResponseWriter, r *http.Request) {
+	ctx, span := trace.StartSpan(r.Context(), "validator.SubmitAggregateAndProofs")
+	defer span.End()
+
+	var req structs.SubmitAggregateAndProofsRequestV2
+	err := json.NewDecoder(r.Body).Decode(&req.Data)
+	switch {
+	case errors.Is(err, io.EOF):
+		httputil.HandleError(w, "No data submitted", http.StatusBadRequest)
+		return
+	case err != nil:
+		httputil.HandleError(w, "Could not decode request body: "+err.Error(), http.StatusBadRequest)
+		return
+	}
+	if len(req.Data) == 0 {
+		httputil.HandleError(w, "No data submitted", http.StatusBadRequest)
+		return
+	}
+
+	versionHeader := r.Header.Get(api.VersionHeader)
+	if versionHeader == "" {
+		httputil.HandleError(w, api.VersionHeader+" header is required", http.StatusBadRequest)
+	}
+	v, err := version.FromString(versionHeader)
+	if err != nil {
+		httputil.HandleError(w, "Invalid version: "+err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	broadcastFailed := false
+	if v >= version.Electra {
+		for _, raw := range req.Data {
+			var item structs.SignedAggregateAttestationAndProofElectra
+			if err = json.Unmarshal(raw, &item); err != nil {
+				httputil.HandleError(w, "Failed to parse Electra aggregate attestation and proof: "+err.Error(), http.StatusBadRequest)
+				return
+			}
+
+			consensusItem, err := item.ToConsensus()
+			if err != nil {
+				httputil.HandleError(w, "Could not convert Electra request aggregate to consensus aggregate: "+err.Error(), http.StatusBadRequest)
+				return
+			}
+
+			rpcError := s.CoreService.SubmitSignedAggregateSelectionProof(ctx, consensusItem)
+			if rpcError != nil {
+				var aggregateBroadcastFailedError *core.AggregateBroadcastFailedError
+				if errors.As(rpcError.Err, &aggregateBroadcastFailedError) {
+					broadcastFailed = true
+				} else {
+					httputil.HandleError(w, rpcError.Err.Error(), core.ErrorReasonToHTTP(rpcError.Reason))
+					return
+				}
+			}
+		}
+	} else {
+		for _, raw := range req.Data {
+			var item structs.SignedAggregateAttestationAndProof
+			if err := json.Unmarshal(raw, &item); err != nil {
+				httputil.HandleError(w, "Failed to parse older version aggregate attestation and proof: "+err.Error(), http.StatusBadRequest)
+				return
+			}
+
+			consensusItem, err := item.ToConsensus()
+			if err != nil {
+				httputil.HandleError(w, "Could not convert request aggregate to consensus aggregate: "+err.Error(), http.StatusBadRequest)
+				return
+			}
+
+			rpcError := s.CoreService.SubmitSignedAggregateSelectionProof(ctx, consensusItem)
+			if rpcError != nil {
+				var aggregateBroadcastFailedError *core.AggregateBroadcastFailedError
+				if errors.As(rpcError.Err, &aggregateBroadcastFailedError) {
+					broadcastFailed = true
+				} else {
+					httputil.HandleError(w, rpcError.Err.Error(), core.ErrorReasonToHTTP(rpcError.Reason))
+					return
+				}
+			}
+		}
+	}
+
+	if broadcastFailed {
+		httputil.HandleError(w, "Could not broadcast one or more signed aggregated attestations", http.StatusInternalServerError)
+	}
+}
+
 // SubmitSyncCommitteeSubscription subscribe to a number of sync committee subnets.
 //
 // Subscribing to sync committee subnets is an action performed by VC to enable

beacon-chain/rpc/eth/validator/handlers_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v5/api"
 	"github.com/prysmaticlabs/prysm/v5/api/server/structs"
 	mockChain "github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain/testing"
 	builderTest "github.com/prysmaticlabs/prysm/v5/beacon-chain/builder/testing"
@@ -37,6 +38,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v5/network/httputil"
 	ethpbalpha "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v5/runtime/version"
 	"github.com/prysmaticlabs/prysm/v5/testing/assert"
 	"github.com/prysmaticlabs/prysm/v5/testing/require"
 	"github.com/prysmaticlabs/prysm/v5/testing/util"
@@ -450,6 +452,22 @@ func TestSubmitAggregateAndProofs(t *testing.T) {
 		assert.Equal(t, http.StatusOK, writer.Code)
 		assert.Equal(t, 1, len(broadcaster.BroadcastMessages))
 	})
+	t.Run("singleV2", func(t *testing.T) {
+		broadcaster := &p2pmock.MockBroadcaster{}
+		c.Broadcaster = broadcaster
+
+		var body bytes.Buffer
+		_, err := body.WriteString(singleAggregate)
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodPost, "http://example.com", &body)
+		request.Header.Set(api.VersionHeader, version.String(version.Electra))
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.SubmitAggregateAndProofsV2(writer, request)
+		assert.Equal(t, http.StatusOK, writer.Code)
+		assert.Equal(t, 1, len(broadcaster.BroadcastMessages))
+	})
 	t.Run("multiple", func(t *testing.T) {
 		broadcaster := &p2pmock.MockBroadcaster{}
 		c.Broadcaster = broadcaster
@@ -466,6 +484,23 @@ func TestSubmitAggregateAndProofs(t *testing.T) {
 		assert.Equal(t, http.StatusOK, writer.Code)
 		assert.Equal(t, 2, len(broadcaster.BroadcastMessages))
 	})
+	t.Run("multipleV2", func(t *testing.T) {
+		broadcaster := &p2pmock.MockBroadcaster{}
+		c.Broadcaster = broadcaster
+		c.SyncCommitteePool = synccommittee.NewStore()
+
+		var body bytes.Buffer
+		_, err := body.WriteString(multipleAggregates)
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodPost, "http://example.com", &body)
+		request.Header.Set(api.VersionHeader, version.String(version.Electra))
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.SubmitAggregateAndProofsV2(writer, request)
+		assert.Equal(t, http.StatusOK, writer.Code)
+		assert.Equal(t, 2, len(broadcaster.BroadcastMessages))
+	})
 	t.Run("no body", func(t *testing.T) {
 		request := httptest.NewRequest(http.MethodPost, "http://example.com", nil)
 		writer := httptest.NewRecorder()
@@ -478,6 +513,19 @@ func TestSubmitAggregateAndProofs(t *testing.T) {
 		assert.Equal(t, http.StatusBadRequest, e.Code)
 		assert.Equal(t, true, strings.Contains(e.Message, "No data submitted"))
 	})
+	t.Run("no bodyV2", func(t *testing.T) {
+		request := httptest.NewRequest(http.MethodPost, "http://example.com", nil)
+		request.Header.Set(api.VersionHeader, version.String(version.Electra))
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.SubmitAggregateAndProofsV2(writer, request)
+		assert.Equal(t, http.StatusBadRequest, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusBadRequest, e.Code)
+		assert.Equal(t, true, strings.Contains(e.Message, "No data submitted"))
+	})
 	t.Run("empty", func(t *testing.T) {
 		var body bytes.Buffer
 		_, err := body.WriteString("[]")
@@ -493,6 +541,22 @@ func TestSubmitAggregateAndProofs(t *testing.T) {
 		assert.Equal(t, http.StatusBadRequest, e.Code)
 		assert.Equal(t, true, strings.Contains(e.Message, "No data submitted"))
 	})
+	t.Run("emptyV2", func(t *testing.T) {
+		var body bytes.Buffer
+		_, err := body.WriteString("[]")
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodPost, "http://example.com", &body)
+		request.Header.Set(api.VersionHeader, version.String(version.Electra))
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.SubmitAggregateAndProofsV2(writer, request)
+		assert.Equal(t, http.StatusBadRequest, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusBadRequest, e.Code)
+		assert.Equal(t, true, strings.Contains(e.Message, "No data submitted"))
+	})
 	t.Run("invalid", func(t *testing.T) {
 		var body bytes.Buffer
 		_, err := body.WriteString(invalidAggregate)
@@ -507,6 +571,21 @@ func TestSubmitAggregateAndProofs(t *testing.T) {
 		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
 		assert.Equal(t, http.StatusBadRequest, e.Code)
 	})
+	t.Run("invalidV2", func(t *testing.T) {
+		var body bytes.Buffer
+		_, err := body.WriteString(invalidAggregate)
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodPost, "http://example.com", &body)
+		request.Header.Set(api.VersionHeader, version.String(version.Electra))
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.SubmitAggregateAndProofsV2(writer, request)
+		assert.Equal(t, http.StatusBadRequest, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusBadRequest, e.Code)
+	})
 }
 
 func TestSubmitSyncCommitteeSubscription(t *testing.T) {
@@ -2668,6 +2747,7 @@ var (
       "aggregator_index": "1",
       "aggregate": {
         "aggregation_bits": "0x01",
+        "committee_bits": "0x01",
         "signature": "0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8bb2305b26a285fa2737f175668d0dff91cc1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505",
         "data": {
           "slot": "1",
@@ -2694,6 +2774,7 @@ var (
       "aggregator_index": "1",
       "aggregate": {
         "aggregation_bits": "0x01",
+		"committee_bits": "0x01",
         "signature": "0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8bb2305b26a285fa2737f175668d0dff91cc1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505",
         "data": {
           "slot": "1",
@@ -2718,6 +2799,7 @@ var (
       "aggregator_index": "1",
       "aggregate": {
         "aggregation_bits": "0x01",
+		"committee_bits": "0x01",
         "signature": "0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8bb2305b26a285fa2737f175668d0dff91cc1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505",
         "data": {
           "slot": "1",
@@ -2746,6 +2828,7 @@ var (
       "aggregator_index": "foo",
       "aggregate": {
         "aggregation_bits": "0x01",
+		"committee_bits": "0x01",
         "signature": "0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8bb2305b26a285fa2737f175668d0dff91cc1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505",
         "data": {
           "slot": "1",