eth/fetcher: validate announced blob tx size against announcer's protocol version (#35524)

eth/72 announces blob transactions without the blob payload while eth/71
includes it, so the same transaction is announced with different
sizes. Compare each announcement against the size expected for that
peer's version instead of a single size so that honest peers on either version
are no longer dropped on delivery.

eth/fetcher/tx_fetcher.go

@@ -34,6 +34,7 @@ import (
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethereum/go-ethereum/params"
 )
 
 const (
@@ -101,8 +102,31 @@ type txAnnounce struct {
 // txMetadata provides the extra data transmitted along with the announcement
 // for better fetch scheduling.
 type txMetadata struct {
-	kind byte   // Transaction consensus type
-	size uint32 // Transaction size in bytes
+	kind    byte   // Transaction consensus type
+	size    uint32 // Transaction size in bytes, as announced
+	version uint   // Protocol version of the announcing peer
+}
+
+// txDeliveryMeta is the metadata of a delivered transaction. eth72 announces
+// blob transactions without the blob payload, so both sizes are kept.
+type txDeliveryMeta struct {
+	kind            byte   // Transaction consensus type
+	size            uint32 // Size with blobs
+	sizeWithoutBlob uint32 // Size without blobs (eth72)
+}
+
+// sizeForVersion returns the size an announcer on the given version advertises.
+func (m *txDeliveryMeta) sizeForVersion(version uint) uint32 {
+	if m.kind == types.BlobTxType && version >= eth.ETH72 {
+		return m.sizeWithoutBlob
+	}
+	return m.size
+}
+
+// blobPayloadSize returns the encoded size of the blob payload omitted (under eth72)
+func blobPayloadSize(n int) uint32 {
+	const blobRLPSize = params.BlobTxFieldElementsPerBlob*params.BlobTxBytesPerFieldElement + 4
+	return uint32(n)*blobRLPSize + 4
 }
 
 // txMetadataWithSeq is a wrapper of transaction metadata with an extra field
@@ -123,11 +147,11 @@ type txRequest struct {
 // txDelivery is the notification that a batch of transactions have been added
 // to the pool and should be untracked.
 type txDelivery struct {
-	origin    string        // Identifier of the peer originating the notification
-	hashes    []common.Hash // Batch of transaction hashes having been delivered
-	metas     []txMetadata  // Batch of metadata associated with the delivered hashes
-	direct    bool          // Whether this is a direct reply or a broadcast
-	violation error         // Whether we encountered a protocol violation
+	origin    string           // Identifier of the peer originating the notification
+	hashes    []common.Hash    // Batch of transaction hashes having been delivered
+	metas     []txDeliveryMeta // Batch of metadata associated with the delivered hashes
+	direct    bool             // Whether this is a direct reply or a broadcast
+	violation error            // Whether we encountered a protocol violation
 }
 
 // txDrop is the notification that a peer has disconnected.
@@ -241,7 +265,7 @@ func NewTxFetcherForTests(
 
 // Notify announces the fetcher of the potential availability of a new batch of
 // transactions in the network. It returns array of hashes decided to be fetched.
-func (f *TxFetcher) Notify(peer string, kinds []byte, sizes []uint32, hashes []common.Hash) ([]common.Hash, error) {
+func (f *TxFetcher) Notify(peer string, version uint, kinds []byte, sizes []uint32, hashes []common.Hash) ([]common.Hash, error) {
 	// Keep track of all the announced transactions
 	txAnnounceInMeter.Mark(int64(len(hashes)))
 
@@ -292,7 +316,7 @@ func (f *TxFetcher) Notify(peer string, kinds []byte, sizes []uint32, hashes []c
 		// Transaction metadata has been available since eth68, and all
 		// legacy eth protocols (prior to eth68) have been deprecated.
 		// Therefore, metadata is always expected in the announcement.
-		unknownMetas = append(unknownMetas, txMetadata{kind: kinds[i], size: sizes[i]})
+		unknownMetas = append(unknownMetas, txMetadata{kind: kinds[i], size: sizes[i], version: version})
 	}
 	txAnnounceKnownMeter.Mark(duplicate)
 	txAnnounceUnderpricedMeter.Mark(underpriced)
@@ -356,7 +380,7 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction,
 	// re-requesting them and dropping the peer in case of malicious transfers.
 	var (
 		added = make([]common.Hash, 0, len(txs))
-		metas = make([]txMetadata, 0, len(txs))
+		metas = make([]txDeliveryMeta, 0, len(txs))
 	)
 	// proceed in batches
 	for i := 0; i < len(txs); i += addTxsBatchSize {
@@ -401,10 +425,21 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction,
 				violation = err
 			}
 			added = append(added, batch[j].Hash())
-			metas = append(metas, txMetadata{
-				kind: batch[j].Type(),
-				size: uint32(batch[j].Size()),
-			})
+			size := uint32(batch[j].Size())
+			meta := txDeliveryMeta{
+				kind:            batch[j].Type(),
+				size:            size,
+				sizeWithoutBlob: size,
+			}
+			if sc := batch[j].BlobTxSidecar(); sc != nil {
+				if version >= eth.ETH72 {
+					// tx should be delivered without blobs
+					meta.size += blobPayloadSize(len(sc.Commitments))
+				} else {
+					meta.sizeWithoutBlob -= blobPayloadSize(len(sc.Commitments))
+				}
+			}
+			metas = append(metas, meta)
 			// Terminate the transaction processing if violation is encountered. All
 			// the remaining transactions in response will be silently discarded.
 			if violation != nil {
@@ -755,9 +790,9 @@ func (f *TxFetcher) loop() {
 							if delivery.metas[i].kind != meta.kind {
 								log.Warn("Announced transaction type mismatch", "peer", peer, "tx", hash, "type", delivery.metas[i].kind, "ann", meta.kind)
 								f.dropPeer(peer)
-							} else if delivery.metas[i].size != meta.size {
-								if math.Abs(float64(delivery.metas[i].size)-float64(meta.size)) > 8 {
-									log.Warn("Announced transaction size mismatch", "peer", peer, "tx", hash, "size", delivery.metas[i].size, "ann", meta.size)
+							} else if size := delivery.metas[i].sizeForVersion(meta.version); size != meta.size {
+								if math.Abs(float64(size)-float64(meta.size)) > 8 {
+									log.Warn("Announced transaction size mismatch", "peer", peer, "tx", hash, "size", size, "ann", meta.size)
 
 									// Normally we should drop a peer considering this is a protocol violation.
 									// However, due to the RLP vs consensus format messyness, allow a few bytes
@@ -781,9 +816,9 @@ func (f *TxFetcher) loop() {
 							if delivery.metas[i].kind != meta.kind {
 								log.Warn("Announced transaction type mismatch", "peer", peer, "tx", hash, "type", delivery.metas[i].kind, "ann", meta.kind)
 								f.dropPeer(peer)
-							} else if delivery.metas[i].size != meta.size {
-								if math.Abs(float64(delivery.metas[i].size)-float64(meta.size)) > 8 {
-									log.Warn("Announced transaction size mismatch", "peer", peer, "tx", hash, "size", delivery.metas[i].size, "ann", meta.size)
+							} else if size := delivery.metas[i].sizeForVersion(meta.version); size != meta.size {
+								if math.Abs(float64(size)-float64(meta.size)) > 8 {
+									log.Warn("Announced transaction size mismatch", "peer", peer, "tx", hash, "size", size, "ann", meta.size)
 
 									// Normally we should drop a peer considering this is a protocol violation.
 									// However, due to the RLP vs consensus format messyness, allow a few bytes

eth/fetcher/tx_fetcher_test.go

@@ -56,10 +56,11 @@ type announce struct {
 }
 
 type doTxNotify struct {
-	peer   string
-	hashes []common.Hash
-	types  []byte
-	sizes  []uint32
+	peer    string
+	version uint
+	hashes  []common.Hash
+	types   []byte
+	sizes   []uint32
 }
 type doTxEnqueue struct {
 	peer    string
@@ -1766,15 +1767,17 @@ func TestTransactionFetcherWrongMetadata(t *testing.T) {
 	})
 }
 
-func makeInvalidBlobTx() *types.Transaction {
+func makeBlobTx(validProof bool) *types.Transaction {
 	key, _ := crypto.GenerateKey()
 	blob := &kzg4844.Blob{byte(0xa)}
 	commitment, _ := kzg4844.BlobToCommitment(blob)
 	blobHash := kzg4844.CalcBlobHashV1(sha256.New(), &commitment)
 	cellProof, _ := kzg4844.ComputeCellProofs(blob)
 
-	// Mutate the cell proof
-	cellProof[0][0] = 0x0
+	if !validProof {
+		// Mutate the cell proof
+		cellProof[0][0] = 0x0
+	}
 
 	blobtx := &types.BlobTx{
 		ChainID:    uint256.MustFromBig(params.MainnetChainConfig.ChainID),
@@ -1796,7 +1799,7 @@ func TestTransactionProtocolViolation(t *testing.T) {
 	//log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelDebug, true)))
 
 	var (
-		badTx = makeInvalidBlobTx()
+		badTx = makeBlobTx(false)
 		drop  = make(chan struct{}, 1)
 	)
 	testTransactionFetcherParallel(t, txFetcherTest{
@@ -1870,6 +1873,44 @@ func TestTransactionProtocolViolation(t *testing.T) {
 	})
 }
 
+// Tests that announced blob transaction sizes are validated against the form
+// matching each announcer's protocol version: full below eth/72, without
+// blobs on eth/72.
+func TestTransactionFetcherBlobSizeVersions(t *testing.T) {
+	var (
+		tx   = makeBlobTx(true)
+		drop = make(chan string, 4)
+	)
+	size := uint32(tx.Size())
+	sizeWithoutBlob := size - blobPayloadSize(len(tx.BlobTxSidecar().Blobs))
+	testTransactionFetcherParallel(t, txFetcherTest{
+		init: func() *TxFetcher {
+			f := newTestTxFetcher()
+			f.dropPeer = func(peer string) { drop <- peer }
+			return f
+		},
+		steps: []interface{}{
+			doTxNotify{peer: "A", version: eth.ETH72, hashes: []common.Hash{tx.Hash()}, types: []byte{types.BlobTxType}, sizes: []uint32{sizeWithoutBlob}},
+			doTxNotify{peer: "B", version: eth.ETH71, hashes: []common.Hash{tx.Hash()}, types: []byte{types.BlobTxType}, sizes: []uint32{size}},
+			doTxNotify{peer: "C", version: eth.ETH72, hashes: []common.Hash{tx.Hash()}, types: []byte{types.BlobTxType}, sizes: []uint32{size}},
+			doWait{time: 0, step: true}, // zero time, but the blob fetching should be scheduled
+
+			// Only C, announcing the wrong form for its version, may be dropped.
+			doTxEnqueue{peer: "B", version: eth.ETH71, txs: []*types.Transaction{tx}, direct: true},
+			doFunc(func() {
+				if peer := <-drop; peer != "C" {
+					t.Fatalf("dropped wrong peer: have %s, want C", peer)
+				}
+				select {
+				case peer := <-drop:
+					t.Fatalf("unexpected peer drop: %s", peer)
+				case <-time.After(10 * time.Millisecond):
+				}
+			}),
+		},
+	})
+}
+
 func testTransactionFetcherParallel(t *testing.T, tt txFetcherTest) {
 	t.Parallel()
 	testTransactionFetcher(t, tt)
@@ -1903,7 +1944,7 @@ func testTransactionFetcher(t *testing.T, tt txFetcherTest) {
 		// Process the original or expanded steps
 		switch step := step.(type) {
 		case doTxNotify:
-			if _, err := fetcher.Notify(step.peer, step.types, step.sizes, step.hashes); err != nil {
+			if _, err := fetcher.Notify(step.peer, step.version, step.types, step.sizes, step.hashes); err != nil {
 				t.Errorf("step %d: %v", i, err)
 			}
 			<-wait // Fetcher needs to process this, wait until it's done

eth/handler_eth.go

@@ -62,7 +62,7 @@ func (h *ethHandler) Handle(peer *eth.Peer, packet eth.Packet) error {
 	// Consume any broadcasts and announces, forwarding the rest to the downloader
 	switch packet := packet.(type) {
 	case *eth.NewPooledTransactionHashesPacket72:
-		hashes, err := h.txFetcher.Notify(peer.ID(), packet.Types, packet.Sizes, packet.Hashes)
+		hashes, err := h.txFetcher.Notify(peer.ID(), peer.Version(), packet.Types, packet.Sizes, packet.Hashes)
 		if err != nil {
 			return err
 		}
@@ -72,7 +72,7 @@ func (h *ethHandler) Handle(peer *eth.Peer, packet eth.Packet) error {
 		return nil
 
 	case *eth.NewPooledTransactionHashesPacket71:
-		_, err := h.txFetcher.Notify(peer.ID(), packet.Types, packet.Sizes, packet.Hashes)
+		_, err := h.txFetcher.Notify(peer.ID(), peer.Version(), packet.Types, packet.Sizes, packet.Hashes)
 		return err
 
 	case *eth.TransactionsPacket:
@@ -134,6 +134,15 @@ func handleTransactions(peer *eth.Peer, list []*types.Transaction, directBroadca
 				if err := tx.BlobTxSidecar().ValidateBlobCommitmentHashes(tx.BlobHashes()); err != nil {
 					return err
 				}
+				// eth72 delivers blob transactions without the blob payload,
+				// earlier versions with all blobs.
+				if blobs := len(tx.BlobTxSidecar().Blobs); peer.Version() >= eth.ETH72 {
+					if blobs != 0 {
+						return errors.New("received blob transaction with blob payload on eth72")
+					}
+				} else if blobs != len(tx.BlobHashes()) {
+					return errors.New("incorrect number of blobs (len(blobs) != len(vhashes))")
+				}
 			}
 		}
 

tests/fuzzers/txfetcher/txfetcher_fuzzer.go

@@ -151,7 +151,7 @@ func fuzz(input []byte) int {
 			if verbose {
 				fmt.Println("Notify", peer, announceIdxs)
 			}
-			if _, err := f.Notify(peer, types, sizes, announces); err != nil {
+			if _, err := f.Notify(peer, peerVersions[peer], types, sizes, announces); err != nil {
 				panic(err)
 			}