From aa1f2fcf512988eb8890d9352e601b898d6fdb2c Mon Sep 17 00:00:00 2001 From: Bosul Mun Date: Thu, 13 Aug 2026 17:42:52 +0200 Subject: [PATCH] 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 | 75 +++++++++++++++------ eth/fetcher/tx_fetcher_test.go | 59 +++++++++++++--- eth/handler_eth.go | 13 +++- tests/fuzzers/txfetcher/txfetcher_fuzzer.go | 2 +- 4 files changed, 117 insertions(+), 32 deletions(-) diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index 63efdfab60..ef7c066768 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/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 diff --git a/eth/fetcher/tx_fetcher_test.go b/eth/fetcher/tx_fetcher_test.go index 1c13c5bfc7..7330dc6a5b 100644 --- a/eth/fetcher/tx_fetcher_test.go +++ b/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 diff --git a/eth/handler_eth.go b/eth/handler_eth.go index 3fae75f3c5..0197773ba9 100644 --- a/eth/handler_eth.go +++ b/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))") + } } } diff --git a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go b/tests/fuzzers/txfetcher/txfetcher_fuzzer.go index 6e9b8acd83..91409b85ec 100644 --- a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go +++ b/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) }