Start validating blobs closer to RPC response reader

beacon-chain/sync/rpc_send_request.go

@@ -25,14 +25,22 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-// ErrInvalidFetchedData is used to signal that an error occurred which should result in peer downscoring.
-var ErrInvalidFetchedData = errors.New("invalid data returned from peer")
-
-var errMaxRequestBlobSidecarsExceeded = errors.Wrap(ErrInvalidFetchedData, "peer exceeded req blob chunk tx limit")
 var errBlobChunkedReadFailure = errors.New("failed to read stream of chunk-encoded blobs")
 var errBlobUnmarshal = errors.New("Could not unmarshal chunk-encoded blob")
-var errUnrequested = errors.New("Received BlobSidecar in response that was not requested")
-var errBlobResponseOutOfBounds = errors.New("received BlobSidecar with slot outside BlobSidecarsByRangeRequest bounds")
+
+// Any error from the following declaration block should result in peer downscoring.
+var (
+	// ErrInvalidFetchedData is used to signal that an error occurred which should result in peer downscoring.
+	ErrInvalidFetchedData             = errors.New("invalid data returned from peer")
+	errBlobIndexOutOfBounds           = errors.Wrap(ErrInvalidFetchedData, "blob index out of range")
+	errMaxRequestBlobSidecarsExceeded = errors.Wrap(ErrInvalidFetchedData, "peer exceeded req blob chunk tx limit")
+	errChunkResponseSlotNotAsc        = errors.Wrap(ErrInvalidFetchedData, "blob slot not higher than previous block root")
+	errChunkResponseIndexNotAsc       = errors.Wrap(ErrInvalidFetchedData, "blob indices for a block must start at 0 and increase by 1")
+	errUnrequested                    = errors.Wrap(ErrInvalidFetchedData, "received BlobSidecar in response that was not requested")
+	errBlobResponseOutOfBounds        = errors.Wrap(ErrInvalidFetchedData, "received BlobSidecar with slot outside BlobSidecarsByRangeRequest bounds")
+	errChunkResponseBlockMismatch     = errors.Wrap(ErrInvalidFetchedData, "blob block details do not match")
+	errChunkResponseParentMismatch    = errors.Wrap(ErrInvalidFetchedData, "parent root for response element doesn't match previous element root")
+)
 
 // BeaconBlockProcessor defines a block processing function, which allows to start utilizing
 // blocks even before all blocks are ready.
@@ -167,7 +175,8 @@ func SendBlobsByRangeRequest(ctx context.Context, tor blockchain.TemporalOracle,
 	if max > req.Count*fieldparams.MaxBlobsPerBlock {
 		max = req.Count * fieldparams.MaxBlobsPerBlock
 	}
-	return readChunkEncodedBlobs(stream, p2pApi.Encoding(), ctxMap, blobValidatorFromRangeReq(req), max)
+	blobVal := composeBlobValidations(blobValidatorFromRangeReq(req), newSequentialBlobValidator())
+	return readChunkEncodedBlobs(stream, p2pApi.Encoding(), ctxMap, blobVal, max)
 }
 
 func SendBlobSidecarByRoot(
@@ -198,6 +207,70 @@ func SendBlobSidecarByRoot(
 
 type blobResponseValidation func(blocks.ROBlob) error
 
+func composeBlobValidations(vf ...blobResponseValidation) blobResponseValidation {
+	return func(blob blocks.ROBlob) error {
+		for i := range vf {
+			if err := vf[i](blob); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
+
+type seqBlobValid struct {
+	prev *blocks.ROBlob
+}
+
+func (sbv *seqBlobValid) nextValid(blob blocks.ROBlob) error {
+	if blob.Index >= fieldparams.MaxBlobsPerBlock {
+		return errBlobIndexOutOfBounds
+	}
+	if sbv.prev == nil {
+		// The first blob we see for a block must have index 0.
+		if blob.Index != 0 {
+			return errChunkResponseIndexNotAsc
+		}
+		sbv.prev = &blob
+		return nil
+	}
+	if sbv.prev.Slot() == blob.Slot() {
+		if sbv.prev.BlockRoot() != blob.BlockRoot() {
+			return errors.Wrap(errChunkResponseBlockMismatch, "block roots do not match")
+		}
+		if sbv.prev.ParentRoot() != blob.ParentRoot() {
+			return errors.Wrap(errChunkResponseBlockMismatch, "block parent roots do not match")
+		}
+		// Blob indices in responses should be strictly monotonically incrementing.
+		if blob.Index != sbv.prev.Index+1 {
+			return errChunkResponseIndexNotAsc
+		}
+	} else {
+		// If the slot is adjacent we know there are no intervening blocks with missing blobs, so we can
+		// check that the new blob descends from the last seen.
+		if blob.Slot() == sbv.prev.Slot()+1 && blob.ParentRoot() != sbv.prev.BlockRoot() {
+			return errChunkResponseParentMismatch
+		}
+		// The first blob we see for a block must have index 0.
+		if blob.Index != 0 {
+			return errChunkResponseIndexNotAsc
+		}
+		// Blocks must be in ascending slot order.
+		if sbv.prev.Slot() >= blob.Slot() {
+			return errChunkResponseSlotNotAsc
+		}
+	}
+	sbv.prev = &blob
+	return nil
+}
+
+func newSequentialBlobValidator() blobResponseValidation {
+	sbv := &seqBlobValid{}
+	return func(blob blocks.ROBlob) error {
+		return sbv.nextValid(blob)
+	}
+}
+
 func blobValidatorFromRootReq(req *p2ptypes.BlobSidecarsByRootReq) blobResponseValidation {
 	blobIds := make(map[[32]byte]map[uint64]bool)
 	for _, sc := range *req {

beacon-chain/sync/rpc_send_request_test.go

@@ -13,6 +13,7 @@ import (
 	p2ptest "github.com/prysmaticlabs/prysm/v4/beacon-chain/p2p/testing"
 	p2pTypes "github.com/prysmaticlabs/prysm/v4/beacon-chain/p2p/types"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/startup"
+	fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/interfaces"
@@ -601,3 +602,88 @@ 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)
+	wrongRoot, err := blocks.NewROBlobWithRoot(oops[2].BlobSidecar, bytesutil.ToBytes32([]byte("parentderp")))
+	require.NoError(t, err)
+	oob := oops[3]
+	oob.Index = fieldparams.MaxBlobsPerBlock
+
+	cases := []struct {
+		name  string
+		seq   []blocks.ROBlob
+		err   error
+		errAt int
+	}{
+		{
+			name: "all valid",
+			seq:  append(append([]blocks.ROBlob{}, oneBlobs...), twoBlobs...),
+		},
+		{
+			name: "idx out of bounds",
+			seq:  []blocks.ROBlob{oob},
+			err:  errBlobIndexOutOfBounds,
+		},
+		{
+			name: "first index is not zero",
+			seq:  []blocks.ROBlob{oneBlobs[1]},
+			err:  errChunkResponseIndexNotAsc,
+		},
+		{
+			name: "index out of order, same block",
+			seq:  []blocks.ROBlob{oneBlobs[1], oneBlobs[0]},
+			err:  errChunkResponseIndexNotAsc,
+		},
+		{
+			name:  "second block starts at idx 1",
+			seq:   []blocks.ROBlob{oneBlobs[0], twoBlobs[1]},
+			err:   errChunkResponseIndexNotAsc,
+			errAt: 1,
+		},
+		{
+			name:  "slots not ascending",
+			seq:   append(append([]blocks.ROBlob{}, twoBlobs...), oops...),
+			err:   errChunkResponseSlotNotAsc,
+			errAt: len(twoBlobs),
+		},
+		{
+			name:  "same slot, different root",
+			seq:   []blocks.ROBlob{oops[0], wrongRoot},
+			err:   errChunkResponseBlockMismatch,
+			errAt: 1,
+		},
+		{
+			name:  "same slot, different parent root",
+			seq:   []blocks.ROBlob{oops[0], oops[1]},
+			err:   errChunkResponseBlockMismatch,
+			errAt: 1,
+		},
+		{
+			name:  "next slot, different parent root",
+			seq:   []blocks.ROBlob{oops[0], twoBlobs[0]},
+			err:   errChunkResponseParentMismatch,
+			errAt: 1,
+		},
+	}
+	for _, c := range cases {
+		sbv := newSequentialBlobValidator()
+		t.Run(c.name, func(t *testing.T) {
+			for i := range c.seq {
+				err := sbv(c.seq[i])
+				if c.err != nil && i == c.errAt {
+					require.ErrorIs(t, err, c.err)
+					return
+				}
+				require.NoError(t, err)
+			}
+		})
+	}
+}