kasey: Option 2

beacon-chain/sync/blobs_test.go

@@ -2,14 +2,10 @@ package sync
 
 import (
 	"context"
-	"encoding/binary"
 	"math"
-	"math/big"
 	"testing"
 	"time"
 
-	"github.com/ethereum/go-ethereum/common"
-	gethTypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/protocol"
 	mock "github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain/testing"
@@ -22,12 +18,10 @@ import (
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
-	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
 	types "github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
 	leakybucket "github.com/prysmaticlabs/prysm/v5/container/leaky-bucket"
 	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v5/network/forks"
-	enginev1 "github.com/prysmaticlabs/prysm/v5/proto/engine/v1"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/testing/require"
 	"github.com/prysmaticlabs/prysm/v5/testing/util"
@@ -57,88 +51,6 @@ type requestFromSidecars func([]blocks.ROBlob) interface{}
 type oldestSlotCallback func(t *testing.T) types.Slot
 type expectedRequirer func(*testing.T, *Service, []*expectedBlobChunk) func(network.Stream)
 
-func generateTestBlockWithSidecars(t *testing.T, parent [32]byte, slot types.Slot, nblobs int) (*ethpb.SignedBeaconBlockDeneb, []blocks.ROBlob) {
-	// Start service with 160 as allowed blocks capacity (and almost zero capacity recovery).
-	stateRoot := bytesutil.PadTo([]byte("stateRoot"), fieldparams.RootLength)
-	receiptsRoot := bytesutil.PadTo([]byte("receiptsRoot"), fieldparams.RootLength)
-	logsBloom := bytesutil.PadTo([]byte("logs"), fieldparams.LogsBloomLength)
-	parentHash := bytesutil.PadTo([]byte("parentHash"), fieldparams.RootLength)
-	tx := gethTypes.NewTransaction(
-		0,
-		common.HexToAddress("095e7baea6a6c7c4c2dfeb977efac326af552d87"),
-		big.NewInt(0), 0, big.NewInt(0),
-		nil,
-	)
-	txs := []*gethTypes.Transaction{tx}
-	encodedBinaryTxs := make([][]byte, 1)
-	var err error
-	encodedBinaryTxs[0], err = txs[0].MarshalBinary()
-	require.NoError(t, err)
-	blockHash := bytesutil.ToBytes32([]byte("foo"))
-	payload := &enginev1.ExecutionPayloadDeneb{
-		ParentHash:    parentHash,
-		FeeRecipient:  make([]byte, fieldparams.FeeRecipientLength),
-		StateRoot:     stateRoot,
-		ReceiptsRoot:  receiptsRoot,
-		LogsBloom:     logsBloom,
-		PrevRandao:    blockHash[:],
-		BlockNumber:   0,
-		GasLimit:      0,
-		GasUsed:       0,
-		Timestamp:     0,
-		ExtraData:     make([]byte, 0),
-		BaseFeePerGas: bytesutil.PadTo([]byte("baseFeePerGas"), fieldparams.RootLength),
-		ExcessBlobGas: 0,
-		BlobGasUsed:   0,
-		BlockHash:     blockHash[:],
-		Transactions:  encodedBinaryTxs,
-	}
-	block := util.NewBeaconBlockDeneb()
-	block.Block.Body.ExecutionPayload = payload
-	block.Block.Slot = slot
-	block.Block.ParentRoot = parent[:]
-	commitments := make([][48]byte, nblobs)
-	block.Block.Body.BlobKzgCommitments = make([][]byte, nblobs)
-	for i := range commitments {
-		binary.LittleEndian.PutUint64(commitments[i][:], uint64(i))
-		block.Block.Body.BlobKzgCommitments[i] = commitments[i][:]
-	}
-
-	root, err := block.Block.HashTreeRoot()
-	require.NoError(t, err)
-
-	sbb, err := blocks.NewSignedBeaconBlock(block)
-	require.NoError(t, err)
-	sidecars := make([]blocks.ROBlob, len(commitments))
-	for i, c := range block.Block.Body.BlobKzgCommitments {
-		sidecars[i] = generateTestSidecar(t, root, sbb, i, c)
-	}
-	return block, sidecars
-}
-
-func generateTestSidecar(t *testing.T, root [32]byte, block interfaces.ReadOnlySignedBeaconBlock, index int, commitment []byte) blocks.ROBlob {
-	header, err := block.Header()
-	require.NoError(t, err)
-	blob := make([]byte, fieldparams.BlobSize)
-	binary.LittleEndian.PutUint64(blob, uint64(index))
-	pb := &ethpb.BlobSidecar{
-		Index:             uint64(index),
-		Blob:              blob,
-		KzgCommitment:     commitment,
-		KzgProof:          commitment,
-		SignedBlockHeader: header,
-	}
-	pb.CommitmentInclusionProof = fakeEmptyProof(t, block, pb)
-
-	sc, err := blocks.NewROBlobWithRoot(pb, root)
-	require.NoError(t, err)
-	return sc
-}
-
-func fakeEmptyProof(_ *testing.T, _ interfaces.ReadOnlySignedBeaconBlock, _ *ethpb.BlobSidecar) [][]byte {
-	return util.HydrateCommitmentInclusionProofs()
-}
-
 type expectedBlobChunk struct {
 	code    uint8
 	sidecar *blocks.ROBlob
@@ -204,12 +116,12 @@ func (c *blobsTestCase) setup(t *testing.T) (*Service, []blocks.ROBlob, func())
 		} else {
 			bs = oldest + types.Slot(i)
 		}
-		block, bsc := generateTestBlockWithSidecars(t, parentRoot, bs, maxBlobs)
-		root, err := block.Block.HashTreeRoot()
-		require.NoError(t, err)
+		block, bsc := util.GenerateTestDenebBlockWithSidecar(t, parentRoot, bs, maxBlobs)
 		sidecars = append(sidecars, bsc...)
-		util.SaveBlock(t, context.Background(), d, block)
-		parentRoot = root
+		pb, err := block.Proto()
+		require.NoError(t, err)
+		util.SaveBlock(t, context.Background(), d, pb)
+		parentRoot = block.Root()
 	}
 
 	client := p2ptest.NewTestP2P(t)

beacon-chain/sync/rpc_beacon_blocks_by_root.go

@@ -139,15 +139,15 @@ func (s *Service) sendAndSaveBlobSidecars(ctx context.Context, request types.Blo
 		return nil
 	}
 
-	sidecars, err := SendBlobSidecarByRoot(ctx, s.cfg.clock, s.cfg.p2p, peerID, s.ctxMap, &request)
-	if err != nil {
-		return err
-	}
-
 	RoBlock, err := blocks.NewROBlock(block)
 	if err != nil {
 		return err
 	}
+	sidecars, err := SendBlobSidecarByRoot(ctx, s.cfg.clock, s.cfg.p2p, peerID, s.ctxMap, &request, s.decoderValidation(RoBlock))
+	if err != nil {
+		return err
+	}
+
 	if len(sidecars) != len(request) {
 		return fmt.Errorf("received %d blob sidecars, expected %d for RPC", len(sidecars), len(request))
 	}

beacon-chain/sync/rpc_beacon_blocks_by_root_test.go

@@ -375,8 +375,7 @@ func TestRequestPendingBlobs(t *testing.T) {
 		require.NoError(t, s.sendAndSaveBlobSidecars(context.Background(), request, "test", b))
 	})
 	t.Run("empty commitment block should not fail", func(t *testing.T) {
-		b, err := blocks.NewSignedBeaconBlock(util.NewBeaconBlock())
-		require.NoError(t, err)
+		b, _ := util.GenerateTestDenebBlockWithSidecar(t, [32]byte{}, 0, 0)
 		request, err := s.pendingBlobsRequestForBlock([32]byte{}, b)
 		require.NoError(t, err)
 		require.NoError(t, s.sendAndSaveBlobSidecars(context.Background(), request, "test", b))
@@ -406,10 +405,7 @@ func TestRequestPendingBlobs(t *testing.T) {
 				blobStorage: filesystem.NewEphemeralBlobStorage(t),
 			},
 		}
-		b := util.NewBeaconBlockDeneb()
-		b.Block.Body.BlobKzgCommitments = make([][]byte, 1)
-		b1, err := blocks.NewSignedBeaconBlock(b)
-		require.NoError(t, err)
+		b1, _ := util.GenerateTestDenebBlockWithSidecar(t, [32]byte{}, 0, 1)
 		request, err := s.pendingBlobsRequestForBlock([32]byte{}, b1)
 		require.NoError(t, err)
 		require.ErrorContains(t, "protocols not supported", s.sendAndSaveBlobSidecars(context.Background(), request, p2.PeerID(), b1))

beacon-chain/sync/rpc_send_request.go

@@ -12,6 +12,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/encoder"
 	p2ptypes "github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p/types"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/verification"
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
@@ -184,7 +185,7 @@ func SendBlobsByRangeRequest(ctx context.Context, tor blockchain.TemporalOracle,
 
 func SendBlobSidecarByRoot(
 	ctx context.Context, tor blockchain.TemporalOracle, p2pApi p2p.P2P, pid peer.ID,
-	ctxMap ContextByteVersions, req *p2ptypes.BlobSidecarsByRootReq,
+	ctxMap ContextByteVersions, req *p2ptypes.BlobSidecarsByRootReq, bvs ...BlobResponseValidation,
 ) ([]blocks.ROBlob, error) {
 	if uint64(len(*req)) > params.BeaconConfig().MaxRequestBlobSidecars {
 		return nil, errors.Wrapf(p2ptypes.ErrMaxBlobReqExceeded, "length=%d", len(*req))
@@ -205,7 +206,11 @@ func SendBlobSidecarByRoot(
 	if max > uint64(len(*req))*fieldparams.MaxBlobsPerBlock {
 		max = uint64(len(*req)) * fieldparams.MaxBlobsPerBlock
 	}
-	return readChunkEncodedBlobs(stream, p2pApi.Encoding(), ctxMap, blobValidatorFromRootReq(req), max)
+	vfuncs := []BlobResponseValidation{blobValidatorFromRootReq(req)}
+	if len(bvs) > 0 {
+		vfuncs = append(vfuncs, bvs...)
+	}
+	return readChunkEncodedBlobs(stream, p2pApi.Encoding(), ctxMap, composeBlobValidations(vfuncs...), max)
 }
 
 // BlobResponseValidation represents a function that can validate aspects of a single unmarshaled blob
@@ -374,3 +379,23 @@ func readChunkedBlobSidecar(stream network.Stream, encoding encoder.NetworkEncod
 
 	return rob, nil
 }
+
+// Save bandwidth by stopping chunked reading as soon as we get an unwanted blob.
+func (s *Service) decoderValidation(block blocks.ROBlock) BlobResponseValidation {
+	return func(sc blocks.ROBlob) error {
+		if block.Signature() != bytesutil.ToBytes96(sc.SignedBlockHeader.Signature) {
+			return verification.ErrInvalidProposerSignature
+		}
+		bv := s.newBlobVerifier(sc, verification.InitsyncSidecarRequirements)
+		if err := bv.BlobIndexInBounds(); err != nil {
+			return err
+		}
+		if err := bv.SidecarInclusionProven(); err != nil {
+			return err
+		}
+		if err := bv.SidecarKzgProofVerified(); err != nil {
+			return err
+		}
+		return nil
+	}
+}

beacon-chain/sync/rpc_send_request_test.go

@@ -604,14 +604,13 @@ func TestBlobValidatorFromRangeReq(t *testing.T) {
 }
 
 func TestSeqBlobValid(t *testing.T) {
-	one, oneBlobs := generateTestBlockWithSidecars(t, [32]byte{}, 0, 3)
-	r1, err := one.Block.HashTreeRoot()
-	require.NoError(t, err)
-	two, twoBlobs := generateTestBlockWithSidecars(t, r1, 1, 3)
-	r2, err := two.Block.HashTreeRoot()
-	require.NoError(t, err)
-	_, oops := generateTestBlockWithSidecars(t, r2, 0, 4)
-	oops[1].SignedBlockHeader.Header.ParentRoot = bytesutil.PadTo([]byte("derp"), 32)
+	one, oneBlobs := util.GenerateTestDenebBlockWithSidecar(t, [32]byte{}, 0, 3)
+	r1 := one.Root()
+	two, twoBlobs := util.GenerateTestDenebBlockWithSidecar(t, r1, 1, 3)
+	r2 := two.Root()
+	_, oops := util.GenerateTestDenebBlockWithSidecar(t, r2, 0, 4)
+	_, wrongParent := util.GenerateTestDenebBlockWithSidecar(t, bytesutil.ToBytes32([]byte("derp")), 0, 4)
+	oops[1] = wrongParent[1]
 	wrongRoot, err := blocks.NewROBlobWithRoot(oops[2].BlobSidecar, bytesutil.ToBytes32([]byte("parentderp")))
 	require.NoError(t, err)
 	oob := oops[3]