fix(node): verify blob content in beacon fallback before accepting a response

node/derivation/base_client_test.go

@@ -9,30 +9,54 @@ import (
 	"sync/atomic"
 	"testing"
 
-	"github.com/morph-l2/go-ethereum/common"
+	"github.com/morph-l2/go-ethereum/common/hexutil"
+	"github.com/morph-l2/go-ethereum/crypto/kzg4844"
 	"github.com/stretchr/testify/require"
 )
 
-// zero-value but correctly-sized hex so the beacon JSON decodes into a
-// BlobSidecar. GetBlobSidecarsEnhanced does not verify blob contents (that
-// happens downstream), it only counts sidecars, so dummy bytes are fine here.
+// GetBlobSidecarsEnhanced authenticates blob content against the requested
+// versioned hashes, so the accept-path stubs must serve a real
+// (blob, commitment) pair. The all-zero blob is the cheapest valid one: its
+// KZG commitment is computed once here and its versioned hash is what tests
+// request.
 var (
 	hex32 = "0x" + strings.Repeat("00", 32)
 	hex48 = "0x" + strings.Repeat("00", 48)
+
+	zeroBlobHex        = "0x" + strings.Repeat("00", BlobSize)
+	zeroBlobCommitment = mustZeroBlobCommitment()
+	zeroBlobHash       = KZGToVersionedHash(zeroBlobCommitment)
+
+	// Valid field elements (first byte 0x01 < BLS modulus high byte) but not
+	// the zero blob, served under the zero blob's commitment: the count and
+	// the commitment lookup both pass, only the KZG round-trip in verifyBlob
+	// can catch it.
+	corruptBlobHex = "0x01" + strings.Repeat("00", BlobSize-1)
 )
 
-func sidecarJSON(index int) string {
-	return fmt.Sprintf(`{"block_root":%q,"slot":"1","blob":"0x00","index":"%d","kzg_commitment":%q,"kzg_proof":%q}`,
-		hex32, index, hex48, hex48)
+func mustZeroBlobCommitment() kzg4844.Commitment {
+	var blob Blob
+	commitment, err := kzg4844.BlobToCommitment(blob.KZGBlob())
+	if err != nil {
+		panic(err)
+	}
+	return commitment
+}
+
+func sidecarJSON(index int, blobHex, commitmentHex string) string {
+	return fmt.Sprintf(`{"block_root":%q,"slot":"1","blob":%q,"index":"%d","kzg_commitment":%q,"kzg_proof":%q}`,
+		hex32, blobHex, index, commitmentHex, hex48)
 }
 
 // beaconBehavior controls what a stub beacon returns for blob_sidecars.
 type beaconBehavior int
 
 const (
-	beaconServesBlob  beaconBehavior = iota // 200 with one sidecar
-	beaconServesEmpty                       // 200 with an empty list (pruned / not indexed)
-	beaconServerError                       // 500
+	beaconServesBlob             beaconBehavior = iota // 200 with one valid sidecar
+	beaconServesEmpty                                  // 200 with an empty list (pruned / not indexed)
+	beaconServerError                                  // 500
+	beaconServesCorruptBlob                            // 200, right count and commitment, blob bytes do not match
+	beaconServesWrongCommitment                        // 200, right count, commitment of some other blob
 )
 
 // newStubBeacon serves the genesis + spec endpoints (needed for slot math) and
@@ -51,11 +75,15 @@ func newStubBeacon(t *testing.T, behavior beaconBehavior) (string, *int32) {
 			atomic.AddInt32(&blobHits, 1)
 			switch behavior {
 			case beaconServesBlob:
-				_, _ = w.Write([]byte(`{"data":[` + sidecarJSON(0) + `]}`))
+				_, _ = w.Write([]byte(`{"data":[` + sidecarJSON(0, zeroBlobHex, hexutil.Encode(zeroBlobCommitment[:])) + `]}`))
 			case beaconServesEmpty:
 				_, _ = w.Write([]byte(`{"data":[]}`))
 			case beaconServerError:
 				w.WriteHeader(http.StatusInternalServerError)
+			case beaconServesCorruptBlob:
+				_, _ = w.Write([]byte(`{"data":[` + sidecarJSON(0, corruptBlobHex, hexutil.Encode(zeroBlobCommitment[:])) + `]}`))
+			case beaconServesWrongCommitment:
+				_, _ = w.Write([]byte(`{"data":[` + sidecarJSON(0, zeroBlobHex, hex48) + `]}`))
 			}
 		default:
 			w.WriteHeader(http.StatusNotFound)
@@ -66,7 +94,7 @@ func newStubBeacon(t *testing.T, behavior beaconBehavior) (string, *int32) {
 }
 
 func oneHash() []IndexedBlobHash {
-	return []IndexedBlobHash{{Index: 0, Hash: common.Hash{}}}
+	return []IndexedBlobHash{{Index: 0, Hash: zeroBlobHash}}
 }
 
 type canceledHTTP struct {
@@ -133,6 +161,37 @@ func TestFallbackBeacon_FallsBackOnTransportError(t *testing.T) {
 	require.Positive(t, atomic.LoadInt32(fallbackHits))
 }
 
+// The primary replies 200 with the right sidecar count and the right
+// commitment, but the blob bytes do not commit to the requested hash
+// (corrupted storage, wrong fork, etc.). Content verification must reject it
+// and fall back instead of handing bad bytes downstream.
+func TestFallbackBeacon_FallsBackOnCorruptBlobContent(t *testing.T) {
+	primary, primaryHits := newStubBeacon(t, beaconServesCorruptBlob)
+	fallback, fallbackHits := newStubBeacon(t, beaconServesBlob)
+
+	c := NewFallbackBeaconClient([]string{primary, fallback}, nil, nil)
+	sidecars, err := fetch(t, c)
+	require.NoError(t, err)
+	require.Len(t, sidecars, 1)
+	require.Positive(t, atomic.LoadInt32(primaryHits))
+	require.Positive(t, atomic.LoadInt32(fallbackHits), "corrupt blob content must trigger fallback")
+}
+
+// The primary replies 200 with the right sidecar count but none of the
+// sidecars carries the requested versioned hash (e.g. sidecars of another
+// block at the same slot during a reorg). Must fall back.
+func TestFallbackBeacon_FallsBackOnMissingRequestedHash(t *testing.T) {
+	primary, primaryHits := newStubBeacon(t, beaconServesWrongCommitment)
+	fallback, fallbackHits := newStubBeacon(t, beaconServesBlob)
+
+	c := NewFallbackBeaconClient([]string{primary, fallback}, nil, nil)
+	sidecars, err := fetch(t, c)
+	require.NoError(t, err)
+	require.Len(t, sidecars, 1)
+	require.Positive(t, atomic.LoadInt32(primaryHits))
+	require.Positive(t, atomic.LoadInt32(fallbackHits), "missing requested hash must trigger fallback")
+}
+
 func TestFallbackBeacon_StopsOnContextCancellation(t *testing.T) {
 	var primaryCalls, fallbackCalls int32
 	c := &FallbackBeaconClient{
@@ -153,11 +212,13 @@ func TestFallbackBeacon_StopsOnContextCancellation(t *testing.T) {
 	require.Zero(t, atomic.LoadInt32(&fallbackCalls))
 }
 
-// When every beacon fails to serve the blob, an error is returned and every
-// endpoint's failure is recorded in metrics (exercised via a real *Metrics).
+// When every beacon fails to serve a valid blob, an error is returned and
+// every endpoint's failure is recorded in metrics (exercised via a real
+// *Metrics). The corrupt-content endpoint proves verification failures are
+// counted the same way as availability failures.
 func TestFallbackBeacon_AllFailReturnsError(t *testing.T) {
 	primary, primaryHits := newStubBeacon(t, beaconServesEmpty)
-	fallback, fallbackHits := newStubBeacon(t, beaconServerError)
+	fallback, fallbackHits := newStubBeacon(t, beaconServesCorruptBlob)
 
 	m := PrometheusMetrics("morphnode_test_" + t.Name())
 	c := NewFallbackBeaconClient([]string{primary, fallback}, nil, m)

node/derivation/beacon.go

@@ -13,6 +13,7 @@ import (
 	"sync"
 
 	"github.com/morph-l2/go-ethereum/common"
+	"github.com/morph-l2/go-ethereum/common/hexutil"
 	"github.com/morph-l2/go-ethereum/core/types"
 	"github.com/morph-l2/go-ethereum/crypto/kzg4844"
 	"github.com/morph-l2/go-ethereum/params"
@@ -210,13 +211,14 @@ func dataAndHashesFromTxs(txs types.Transactions, targetTx *types.Transaction) [
 }
 
 // FallbackBeaconClient queries several beacon nodes in order for blob sidecars.
-// A beacon is skipped and the next one tried when it errors, is unreachable, or
-// answers with too few sidecars — a beacon that pruned the slot or has not
-// indexed it yet still replies 200 with an empty/partial list, which is exactly
-// the "temporarily failed to fetch blob" case seen in production and which a
-// transport-level fallback would miss. The failing endpoint is recorded in the
-// beacon_request_failure_total metric so a flaky node is visible on dashboards.
-// With a single endpoint it behaves like a bare L1BeaconClient.
+// A beacon is skipped and the next one tried when it errors, is unreachable,
+// answers with too few sidecars, or serves blob bytes that do not commit to
+// the requested versioned hashes — a beacon that pruned the slot, has not
+// indexed it yet, or holds corrupted blob data still replies 200 with a
+// plausible-looking list, which a transport-level fallback would miss. The
+// failing endpoint is recorded in the beacon_request_failure_total metric so
+// a flaky node is visible on dashboards. With a single endpoint it behaves
+// like a bare L1BeaconClient.
 type FallbackBeaconClient struct {
 	clients   []*L1BeaconClient
 	endpoints []string // parallel to clients, used only for logs/metrics
@@ -237,36 +239,85 @@ func NewFallbackBeaconClient(endpoints []string, log tmlog.Logger, metrics *Metr
 	}
 }
 
-// GetBlobSidecarsEnhanced tries each configured beacon in order and returns the
-// first response that actually carries the requested blobs (at least len(hashes)
-// sidecars, or at least one when no explicit hashes are requested). A beacon
-// that errors or returns an incomplete set is recorded as a failure and skipped.
-// If every beacon fails, the last error is returned.
+// GetBlobSidecarsEnhanced tries each configured beacon in order and returns
+// the first response that actually carries all requested blobs: at least
+// len(hashes) sidecars (at least one when no explicit hashes are requested),
+// every requested versioned hash present among them, and each matched blob's
+// bytes authenticated against its hash via verifySidecars. A beacon that
+// errors, returns an incomplete set, or serves invalid blob content is
+// recorded as a failure and skipped, so callers never receive unverified
+// data while a healthy fallback exists. If every beacon fails, the last
+// error is returned.
 func (c *FallbackBeaconClient) GetBlobSidecarsEnhanced(ctx context.Context, ref L1BlockRef, hashes []IndexedBlobHash) ([]*BlobSidecar, error) {
 	var lastErr error
 	for i, cl := range c.clients {
 		sidecars, err := cl.GetBlobSidecarsEnhanced(ctx, ref, hashes)
-		if err == nil && len(sidecars) > 0 && len(sidecars) >= len(hashes) {
-			return sidecars, nil
-		}
-		if err := ctx.Err(); err != nil {
-			return nil, err
+		if err == nil {
+			if len(sidecars) == 0 || len(sidecars) < len(hashes) {
+				err = fmt.Errorf("beacon returned %d sidecars, want at least %d", len(sidecars), len(hashes))
+			} else {
+				err = verifySidecars(sidecars, hashes)
+			}
 		}
 		if err == nil {
-			err = fmt.Errorf("beacon returned %d sidecars, want at least %d", len(sidecars), len(hashes))
+			return sidecars, nil
+		}
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return nil, ctxErr
 		}
 		if c.metrics != nil {
 			c.metrics.IncBeaconRequestFailure(c.endpoints[i])
 		}
 		if c.log != nil {
-			c.log.Error("beacon failed to serve blob sidecars, trying next endpoint",
+			c.log.Error("beacon failed to serve valid blob sidecars, trying next endpoint",
 				"endpoint", c.endpoints[i], "err", err)
 		}
 		lastErr = err
 	}
 	return nil, lastErr
 }
 
+// verifySidecars checks that, for every requested versioned hash, sidecars
+// contains a blob whose bytes actually commit to that hash. Sidecars are
+// matched by the versioned hash derived from their beacon-supplied
+// commitment, then authenticated with verifyBlob's local KZG commitment
+// round-trip, so a beacon can neither omit a requested blob nor serve
+// corrupted bytes under a correct-looking commitment. Extra sidecars (e.g.
+// from a beacon that ignores the indices filter) are tolerated and simply
+// not inspected. With no requested hashes there is nothing to authenticate
+// and the response is accepted as-is.
+func verifySidecars(sidecars []*BlobSidecar, hashes []IndexedBlobHash) error {
+	if len(hashes) == 0 {
+		return nil
+	}
+	byHash := make(map[common.Hash]*BlobSidecar, len(sidecars))
+	for _, sidecar := range sidecars {
+		var commitment kzg4844.Commitment
+		copy(commitment[:], sidecar.KZGCommitment[:])
+		byHash[KZGToVersionedHash(commitment)] = sidecar
+	}
+	for i := range hashes {
+		expected := hashes[i].Hash
+		sidecar, ok := byHash[expected]
+		if !ok {
+			return fmt.Errorf("blob (hash=%s) not found in beacon response", expected.Hex())
+		}
+		b, err := hexutil.Decode(sidecar.Blob)
+		if err != nil {
+			return fmt.Errorf("failed to decode blob (hash=%s): %w", expected.Hex(), err)
+		}
+		if len(b) != BlobSize {
+			return fmt.Errorf("blob (hash=%s): unexpected length %d (want %d)", expected.Hex(), len(b), BlobSize)
+		}
+		var blob Blob
+		copy(blob[:], b)
+		if err := verifyBlob(&blob, expected); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // Note: ForceGetAllBlobs is defined in derivation.go in the same package
 
 // GetBlobSidecarsEnhanced is an enhanced version of GetBlobSidecars method, combining two approaches to fetch blob data

node/derivation/derivation.go

@@ -565,22 +565,22 @@ func (d *Derivation) fetchRollupDataByTxHash(txHash common.Hash, blockNumber uin
 	if len(blobHashes) > 0 {
 		d.logger.Info("Transaction contains blobs", "txHash", txHash, "blobCount", len(blobHashes))
 
-		// Initialize indexedBlobHashes as nil
-		var indexedBlobHashes []IndexedBlobHash
-
-		// Only try to build IndexedBlobHash array if not forcing get all blobs
-		// Try to get the block to build IndexedBlobHash array
+		// The block body is required to compute each blob's index within the
+		// block's sidecar list, and the resulting IndexedBlobHash set is what
+		// lets the beacon fallback verify blob content per endpoint. A fetch
+		// failure here is treated as fatal for this attempt: the caller's
+		// poll loop retries the whole batch next round, which is preferable
+		// to querying beacons with no hashes to authenticate against.
 		block, err := d.l1Client.BlockByNumber(d.ctx, big.NewInt(int64(blockNumber)))
-		if err == nil {
-			// Successfully got the block, now build IndexedBlobHash array
-			d.logger.Info("Building IndexedBlobHash array from block", "blockNumber", blockNumber)
-			indexedBlobHashes = dataAndHashesFromTxs(block.Transactions(), tx)
-			d.logger.Info("Built IndexedBlobHash array", "count", len(indexedBlobHashes))
-		} else {
-			d.logger.Info("Failed to get block, will try fetching all blobs", "blockNumber", blockNumber, "error", err)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get block %d for blob indices: %w", blockNumber, err)
 		}
+		indexedBlobHashes := dataAndHashesFromTxs(block.Transactions(), tx)
+		d.logger.Info("Built IndexedBlobHash array", "count", len(indexedBlobHashes))
 
-		// Get all blobs corresponding to this timestamp
+		// Fetch the batch's blobs; the fallback client only returns sidecars
+		// whose content already verified against indexedBlobHashes, rotating
+		// to the next beacon endpoint on any invalid response.
 		blobSidecars, err := d.l1BeaconClient.GetBlobSidecarsEnhanced(d.ctx, L1BlockRef{
 			Time: header.Time,
 		}, indexedBlobHashes)