Merge branch 'ethereum:master' into gethintegration

cmd/devp2p/internal/ethtest/conn.go

@@ -130,11 +130,16 @@ func (c *Conn) Write(proto Proto, code uint64, msg any) error {
 	return err
 }
 
+var errDisc error = fmt.Errorf("disconnect")
+
 // ReadEth reads an Eth sub-protocol wire message.
 func (c *Conn) ReadEth() (any, error) {
 	c.SetReadDeadline(time.Now().Add(timeout))
 	for {
 		code, data, _, err := c.Conn.Read()
+		if code == discMsg {
+			return nil, errDisc
+		}
 		if err != nil {
 			return nil, err
 		}

cmd/devp2p/internal/ethtest/suite.go

@@ -17,8 +17,12 @@
 package ethtest
 
 import (
+	"context"
 	"crypto/rand"
+	"fmt"
 	"reflect"
+	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus/misc/eip4844"
@@ -79,6 +83,8 @@ func (s *Suite) EthTests() []utesting.Test {
 		{Name: "InvalidTxs", Fn: s.TestInvalidTxs},
 		{Name: "NewPooledTxs", Fn: s.TestNewPooledTxs},
 		{Name: "BlobViolations", Fn: s.TestBlobViolations},
+		{Name: "TestBlobTxWithoutSidecar", Fn: s.TestBlobTxWithoutSidecar},
+		{Name: "TestBlobTxWithMismatchedSidecar", Fn: s.TestBlobTxWithMismatchedSidecar},
 	}
 }
 
@@ -825,3 +831,194 @@ func (s *Suite) TestBlobViolations(t *utesting.T) {
 		conn.Close()
 	}
 }
+
+// mangleSidecar returns a copy of the given blob transaction where the sidecar
+// data has been modified to produce a different commitment hash.
+func mangleSidecar(tx *types.Transaction) *types.Transaction {
+	sidecar := tx.BlobTxSidecar()
+	copy := types.BlobTxSidecar{
+		Blobs:       append([]kzg4844.Blob{}, sidecar.Blobs...),
+		Commitments: append([]kzg4844.Commitment{}, sidecar.Commitments...),
+		Proofs:      append([]kzg4844.Proof{}, sidecar.Proofs...),
+	}
+	// zero the first commitment to alter the sidecar hash
+	copy.Commitments[0] = kzg4844.Commitment{}
+	return tx.WithBlobTxSidecar(&copy)
+}
+
+func (s *Suite) TestBlobTxWithoutSidecar(t *utesting.T) {
+	t.Log(`This test checks that a blob transaction first advertised/transmitted without blobs will result in the sending peer being disconnected, and the full transaction should be successfully retrieved from another peer.`)
+	tx := s.makeBlobTxs(1, 2, 42)[0]
+	badTx := tx.WithoutBlobTxSidecar()
+	s.testBadBlobTx(t, tx, badTx)
+}
+
+func (s *Suite) TestBlobTxWithMismatchedSidecar(t *utesting.T) {
+	t.Log(`This test checks that a blob transaction first advertised/transmitted without blobs, whose commitment don't correspond to the blob_versioned_hashes in the transaction, will result in the sending peer being disconnected, and the full transaction should be successfully retrieved from another peer.`)
+	tx := s.makeBlobTxs(1, 2, 43)[0]
+	badTx := mangleSidecar(tx)
+	s.testBadBlobTx(t, tx, badTx)
+}
+
+// readUntil reads eth protocol messages until a message of the target type is
+// received.  It returns an error if there is a disconnect, or if the context
+// is cancelled before a message of the desired type can be read.
+func readUntil[T any](ctx context.Context, conn *Conn) (*T, error) {
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, context.Canceled
+		default:
+		}
+		received, err := conn.ReadEth()
+		if err != nil {
+			if err == errDisc {
+				return nil, errDisc
+			}
+			continue
+		}
+
+		switch res := received.(type) {
+		case *T:
+			return res, nil
+		}
+	}
+}
+
+// readUntilDisconnect reads eth protocol messages until the peer disconnects.
+// It returns whether the peer disconnects in the next 100ms.
+func readUntilDisconnect(conn *Conn) (disconnected bool) {
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	_, err := readUntil[struct{}](ctx, conn)
+	return err == errDisc
+}
+
+func (s *Suite) testBadBlobTx(t *utesting.T, tx *types.Transaction, badTx *types.Transaction) {
+	stage1, stage2, stage3 := new(sync.WaitGroup), new(sync.WaitGroup), new(sync.WaitGroup)
+	stage1.Add(1)
+	stage2.Add(1)
+	stage3.Add(1)
+
+	errc := make(chan error)
+
+	badPeer := func() {
+		// announce the correct hash from the bad peer.
+		// when the transaction is first requested before transmitting it from the bad peer,
+		// trigger step 2: connection and announcement by good peers
+
+		conn, err := s.dial()
+		if err != nil {
+			errc <- fmt.Errorf("dial fail: %v", err)
+			return
+		}
+		defer conn.Close()
+
+		if err := conn.peer(s.chain, nil); err != nil {
+			errc <- fmt.Errorf("bad peer: peering failed: %v", err)
+			return
+		}
+
+		ann := eth.NewPooledTransactionHashesPacket{
+			Types:  []byte{types.BlobTxType},
+			Sizes:  []uint32{uint32(badTx.Size())},
+			Hashes: []common.Hash{badTx.Hash()},
+		}
+
+		if err := conn.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil {
+			errc <- fmt.Errorf("sending announcement failed: %v", err)
+			return
+		}
+
+		req, err := readUntil[eth.GetPooledTransactionsPacket](context.Background(), conn)
+		if err != nil {
+			errc <- fmt.Errorf("failed to read GetPooledTransactions message: %v", err)
+			return
+		}
+
+		stage1.Done()
+		stage2.Wait()
+
+		// the good peer is connected, and has announced the tx.
+		// proceed to send the incorrect one from the bad peer.
+
+		resp := eth.PooledTransactionsPacket{RequestId: req.RequestId, PooledTransactionsResponse: eth.PooledTransactionsResponse(types.Transactions{badTx})}
+		if err := conn.Write(ethProto, eth.PooledTransactionsMsg, resp); err != nil {
+			errc <- fmt.Errorf("writing pooled tx response failed: %v", err)
+			return
+		}
+		if !readUntilDisconnect(conn) {
+			errc <- fmt.Errorf("expected bad peer to be disconnected")
+			return
+		}
+		stage3.Done()
+	}
+
+	goodPeer := func() {
+		stage1.Wait()
+
+		conn, err := s.dial()
+		if err != nil {
+			errc <- fmt.Errorf("dial fail: %v", err)
+			return
+		}
+		defer conn.Close()
+
+		if err := conn.peer(s.chain, nil); err != nil {
+			errc <- fmt.Errorf("peering failed: %v", err)
+			return
+		}
+
+		ann := eth.NewPooledTransactionHashesPacket{
+			Types:  []byte{types.BlobTxType},
+			Sizes:  []uint32{uint32(tx.Size())},
+			Hashes: []common.Hash{tx.Hash()},
+		}
+
+		if err := conn.Write(ethProto, eth.NewPooledTransactionHashesMsg, ann); err != nil {
+			errc <- fmt.Errorf("sending announcement failed: %v", err)
+			return
+		}
+
+		// wait until the bad peer has transmitted the incorrect transaction
+		stage2.Done()
+		stage3.Wait()
+
+		// the bad peer has transmitted the bad tx, and been disconnected.
+		// transmit the same tx but with correct sidecar from the good peer.
+
+		var req *eth.GetPooledTransactionsPacket
+		req, err = readUntil[eth.GetPooledTransactionsPacket](context.Background(), conn)
+		if err != nil {
+			errc <- fmt.Errorf("reading pooled tx request failed: %v", err)
+			return
+		}
+
+		if req.GetPooledTransactionsRequest[0] != tx.Hash() {
+			errc <- fmt.Errorf("requested unknown tx hash")
+			return
+		}
+
+		resp := eth.PooledTransactionsPacket{RequestId: req.RequestId, PooledTransactionsResponse: eth.PooledTransactionsResponse(types.Transactions{tx})}
+		if err := conn.Write(ethProto, eth.PooledTransactionsMsg, resp); err != nil {
+			errc <- fmt.Errorf("writing pooled tx response failed: %v", err)
+			return
+		}
+		if readUntilDisconnect(conn) {
+			errc <- fmt.Errorf("unexpected disconnect")
+			return
+		}
+		close(errc)
+	}
+
+	if err := s.engine.sendForkchoiceUpdated(); err != nil {
+		t.Fatalf("send fcu failed: %v", err)
+	}
+
+	go goodPeer()
+	go badPeer()
+	err := <-errc
+	if err != nil {
+		t.Fatalf("%v", err)
+	}
+}

core/txpool/blobpool/blobpool.go

@@ -1085,19 +1085,24 @@ func (p *BlobPool) SetGasTip(tip *big.Int) {
 	p.updateStorageMetrics()
 }
 
-// validateTx checks whether a transaction is valid according to the consensus
-// rules and adheres to some heuristic limits of the local node (price and size).
-func (p *BlobPool) validateTx(tx *types.Transaction) error {
-	// Ensure the transaction adheres to basic pool filters (type, size, tip) and
-	// consensus rules
-	baseOpts := &txpool.ValidationOptions{
+// ValidateTxBasics checks whether a transaction is valid according to the consensus
+// rules, but does not check state-dependent validation such as sufficient balance.
+// This check is meant as an early check which only needs to be performed once,
+// and does not require the pool mutex to be held.
+func (p *BlobPool) ValidateTxBasics(tx *types.Transaction) error {
+	opts := &txpool.ValidationOptions{
 		Config:  p.chain.Config(),
 		Accept:  1 << types.BlobTxType,
 		MaxSize: txMaxSize,
 		MinTip:  p.gasTip.ToBig(),
 	}
+	return txpool.ValidateTransaction(tx, p.head, p.signer, opts)
+}
 
-	if err := p.txValidationFn(tx, p.head, p.signer, baseOpts); err != nil {
+// validateTx checks whether a transaction is valid according to the consensus
+// rules and adheres to some heuristic limits of the local node (price and size).
+func (p *BlobPool) validateTx(tx *types.Transaction) error {
+	if err := p.ValidateTxBasics(tx); err != nil {
 		return err
 	}
 	// Ensure the transaction adheres to the stateful pool filters (nonce, balance)

core/txpool/legacypool/legacypool.go

@@ -558,11 +558,11 @@ func (pool *LegacyPool) Pending(filter txpool.PendingFilter) map[common.Address]
 	return pending
 }
 
-// validateTxBasics checks whether a transaction is valid according to the consensus
+// ValidateTxBasics checks whether a transaction is valid according to the consensus
 // rules, but does not check state-dependent validation such as sufficient balance.
 // This check is meant as an early check which only needs to be performed once,
 // and does not require the pool mutex to be held.
-func (pool *LegacyPool) validateTxBasics(tx *types.Transaction) error {
+func (pool *LegacyPool) ValidateTxBasics(tx *types.Transaction) error {
 	opts := &txpool.ValidationOptions{
 		Config: pool.chainconfig,
 		Accept: 0 |
@@ -573,10 +573,7 @@ func (pool *LegacyPool) validateTxBasics(tx *types.Transaction) error {
 		MaxSize: txMaxSize,
 		MinTip:  pool.gasTip.Load().ToBig(),
 	}
-	if err := txpool.ValidateTransaction(tx, pool.currentHead.Load(), pool.signer, opts); err != nil {
-		return err
-	}
-	return nil
+	return txpool.ValidateTransaction(tx, pool.currentHead.Load(), pool.signer, opts)
 }
 
 // validateTx checks whether a transaction is valid according to the consensus
@@ -929,7 +926,7 @@ func (pool *LegacyPool) Add(txs []*types.Transaction, sync bool) []error {
 		// Exclude transactions with basic errors, e.g invalid signatures and
 		// insufficient intrinsic gas as soon as possible and cache senders
 		// in transactions before obtaining lock
-		if err := pool.validateTxBasics(tx); err != nil {
+		if err := pool.ValidateTxBasics(tx); err != nil {
 			errs[i] = err
 			log.Trace("Discarding invalid transaction", "hash", tx.Hash(), "err", err)
 			invalidTxMeter.Mark(1)

core/txpool/locals/tx_tracker.go

@@ -74,28 +74,45 @@ func New(journalPath string, journalTime time.Duration, chainConfig *params.Chai
 
 // Track adds a transaction to the tracked set.
 // Note: blob-type transactions are ignored.
-func (tracker *TxTracker) Track(tx *types.Transaction) {
-	tracker.TrackAll([]*types.Transaction{tx})
+func (tracker *TxTracker) Track(tx *types.Transaction) error {
+	return tracker.TrackAll([]*types.Transaction{tx})[0]
 }
 
 // TrackAll adds a list of transactions to the tracked set.
 // Note: blob-type transactions are ignored.
-func (tracker *TxTracker) TrackAll(txs []*types.Transaction) {
+func (tracker *TxTracker) TrackAll(txs []*types.Transaction) []error {
 	tracker.mu.Lock()
 	defer tracker.mu.Unlock()
 
+	var errors []error
 	for _, tx := range txs {
 		if tx.Type() == types.BlobTxType {
+			errors = append(errors, nil)
+			continue
+		}
+		// Ignore the transactions which are failed for fundamental
+		// validation such as invalid parameters.
+		if err := tracker.pool.ValidateTxBasics(tx); err != nil {
+			log.Debug("Invalid transaction submitted", "hash", tx.Hash(), "err", err)
+			errors = append(errors, err)
 			continue
 		}
 		// If we're already tracking it, it's a no-op
 		if _, ok := tracker.all[tx.Hash()]; ok {
+			errors = append(errors, nil)
 			continue
 		}
+		// Theoretically, checking the error here is unnecessary since sender recovery
+		// is already part of basic validation. However, retrieving the sender address
+		// from the transaction cache is effectively a no-op if it was previously verified.
+		// Therefore, the error is still checked just in case.
 		addr, err := types.Sender(tracker.signer, tx)
-		if err != nil { // Ignore this tx
+		if err != nil {
+			errors = append(errors, err)
 			continue
 		}
+		errors = append(errors, nil)
+
 		tracker.all[tx.Hash()] = tx
 		if tracker.byAddr[addr] == nil {
 			tracker.byAddr[addr] = legacypool.NewSortedMap()
@@ -107,6 +124,7 @@ func (tracker *TxTracker) TrackAll(txs []*types.Transaction) {
 		}
 	}
 	localGauge.Update(int64(len(tracker.all)))
+	return errors
 }
 
 // recheck checks and returns any transactions that needs to be resubmitted.

core/txpool/subpool.go

@@ -129,6 +129,12 @@ type SubPool interface {
 	// retrieve blobs from the pools directly instead of the network.
 	GetBlobs(vhashes []common.Hash) ([]*kzg4844.Blob, []*kzg4844.Proof)
 
+	// ValidateTxBasics checks whether a transaction is valid according to the consensus
+	// rules, but does not check state-dependent validation such as sufficient balance.
+	// This check is meant as a static check which can be performed without holding the
+	// pool mutex.
+	ValidateTxBasics(tx *types.Transaction) error
+
 	// Add enqueues a batch of transactions into the pool if they are valid. Due
 	// to the large transaction churn, add may postpone fully integrating the tx
 	// to a later point to batch multiple ones together.

core/txpool/txpool.go

@@ -325,6 +325,17 @@ func (p *TxPool) GetBlobs(vhashes []common.Hash) ([]*kzg4844.Blob, []*kzg4844.Pr
 	return nil, nil
 }
 
+// ValidateTxBasics checks whether a transaction is valid according to the consensus
+// rules, but does not check state-dependent validation such as sufficient balance.
+func (p *TxPool) ValidateTxBasics(tx *types.Transaction) error {
+	for _, subpool := range p.subpools {
+		if subpool.Filter(tx) {
+			return subpool.ValidateTxBasics(tx)
+		}
+	}
+	return fmt.Errorf("%w: received type %d", core.ErrTxTypeNotSupported, tx.Type())
+}
+
 // Add enqueues a batch of transactions into the pool if they are valid. Due
 // to the large transaction churn, add may postpone fully integrating the tx
 // to a later point to batch multiple ones together.

core/txpool/validation.go

@@ -17,7 +17,6 @@
 package txpool
 
 import (
-	"crypto/sha256"
 	"errors"
 	"fmt"
 	"math/big"
@@ -170,20 +169,11 @@ func validateBlobSidecar(hashes []common.Hash, sidecar *types.BlobTxSidecar) err
 	if len(sidecar.Blobs) != len(hashes) {
 		return fmt.Errorf("invalid number of %d blobs compared to %d blob hashes", len(sidecar.Blobs), len(hashes))
 	}
-	if len(sidecar.Commitments) != len(hashes) {
-		return fmt.Errorf("invalid number of %d blob commitments compared to %d blob hashes", len(sidecar.Commitments), len(hashes))
-	}
 	if len(sidecar.Proofs) != len(hashes) {
 		return fmt.Errorf("invalid number of %d blob proofs compared to %d blob hashes", len(sidecar.Proofs), len(hashes))
 	}
-	// Blob quantities match up, validate that the provers match with the
-	// transaction hash before getting to the cryptography
-	hasher := sha256.New()
-	for i, vhash := range hashes {
-		computed := kzg4844.CalcBlobHashV1(hasher, &sidecar.Commitments[i])
-		if vhash != computed {
-			return fmt.Errorf("blob %d: computed hash %#x mismatches transaction one %#x", i, computed, vhash)
-		}
+	if err := sidecar.ValidateBlobCommitmentHashes(hashes); err != nil {
+		return err
 	}
 	// Blob commitments match with the hashes in the transaction, verify the
 	// blobs themselves via KZG

core/types/tx_blob.go

@@ -19,6 +19,7 @@ package types
 import (
 	"bytes"
 	"crypto/sha256"
+	"fmt"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -85,6 +86,22 @@ func (sc *BlobTxSidecar) encodedSize() uint64 {
 	return rlp.ListSize(blobs) + rlp.ListSize(commitments) + rlp.ListSize(proofs)
 }
 
+// ValidateBlobCommitmentHashes checks whether the given hashes correspond to the
+// commitments in the sidecar
+func (sc *BlobTxSidecar) ValidateBlobCommitmentHashes(hashes []common.Hash) error {
+	if len(sc.Commitments) != len(hashes) {
+		return fmt.Errorf("invalid number of %d blob commitments compared to %d blob hashes", len(sc.Commitments), len(hashes))
+	}
+	hasher := sha256.New()
+	for i, vhash := range hashes {
+		computed := kzg4844.CalcBlobHashV1(hasher, &sc.Commitments[i])
+		if vhash != computed {
+			return fmt.Errorf("blob %d: computed hash %#x mismatches transaction one %#x", i, computed, vhash)
+		}
+	}
+	return nil
+}
+
 // blobTxWithBlobs is used for encoding of transactions when blobs are present.
 type blobTxWithBlobs struct {
 	BlobTx      *BlobTx

eth/api_backend.go

@@ -272,10 +272,20 @@ func (b *EthAPIBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscri
 }
 
 func (b *EthAPIBackend) SendTx(ctx context.Context, signedTx *types.Transaction) error {
-	if locals := b.eth.localTxTracker; locals != nil {
-		locals.Track(signedTx)
+	locals := b.eth.localTxTracker
+	if locals != nil {
+		if err := locals.Track(signedTx); err != nil {
+			return err
 		}
-	return b.eth.txPool.Add([]*types.Transaction{signedTx}, false)[0]
+	}
+	// No error will be returned to user if the transaction fails stateful
+	// validation (e.g., no available slot), as the locally submitted transactions
+	// may be resubmitted later via the local tracker.
+	err := b.eth.txPool.Add([]*types.Transaction{signedTx}, false)[0]
+	if err != nil && locals == nil {
+		return err
+	}
+	return nil
 }
 
 func (b *EthAPIBackend) GetPoolTransactions() (types.Transactions, error) {

eth/handler_eth.go

@@ -69,6 +69,19 @@ func (h *ethHandler) Handle(peer *eth.Peer, packet eth.Packet) error {
 		return h.txFetcher.Enqueue(peer.ID(), *packet, false)
 
 	case *eth.PooledTransactionsResponse:
+		// If we receive any blob transactions missing sidecars, or with
+		// sidecars that don't correspond to the versioned hashes reported
+		// in the header, disconnect from the sending peer.
+		for _, tx := range *packet {
+			if tx.Type() == types.BlobTxType {
+				if tx.BlobTxSidecar() == nil {
+					return errors.New("received sidecar-less blob transaction")
+				}
+				if err := tx.BlobTxSidecar().ValidateBlobCommitmentHashes(tx.BlobHashes()); err != nil {
+					return err
+				}
+			}
+		}
 		return h.txFetcher.Enqueue(peer.ID(), *packet, true)
 
 	default:

ethclient/ethclient.go

@@ -598,6 +598,15 @@ func (ec *Client) SuggestGasTipCap(ctx context.Context) (*big.Int, error) {
 	return (*big.Int)(&hex), nil
 }
 
+// BlobBaseFee retrieves the current blob base fee.
+func (ec *Client) BlobBaseFee(ctx context.Context) (*big.Int, error) {
+	var hex hexutil.Big
+	if err := ec.c.CallContext(ctx, &hex, "eth_blobBaseFee"); err != nil {
+		return nil, err
+	}
+	return (*big.Int)(&hex), nil
+}
+
 type feeHistoryResultMarshaling struct {
 	OldestBlock  *hexutil.Big     `json:"oldestBlock"`
 	Reward       [][]*hexutil.Big `json:"reward,omitempty"`

ethclient/ethclient_test.go

@@ -404,6 +404,15 @@ func testStatusFunctions(t *testing.T, client *rpc.Client) {
 		t.Fatalf("unexpected gas tip cap: %v", gasTipCap)
 	}
 
+	// BlobBaseFee
+	blobBaseFee, err := ec.BlobBaseFee(context.Background())
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if blobBaseFee.Cmp(big.NewInt(1)) != 0 {
+		t.Fatalf("unexpected blob base fee: %v", blobBaseFee)
+	}
+
 	// FeeHistory
 	history, err := ec.FeeHistory(context.Background(), 1, big.NewInt(2), []float64{95, 99})
 	if err != nil {

version/version.go

@@ -19,6 +19,6 @@ package version
 const (
 	Major = 1          // Major version component of the current release
 	Minor = 15         // Minor version component of the current release
-	Patch = 4          // Patch version component of the current release
+	Patch = 5          // Patch version component of the current release
 	Meta  = "unstable" // Version metadata to append to the version string
 )