Squashed commit of the following:

commit 4b9c307
Author: Amir Hossein <77993374+Kamandlou@users.noreply.github.com>
Date:   Fri Sep 2 20:37:55 2022 +0430

    trie: fix unhandled error in test (ethereum#25628)

commit 0d68b6b
Author: Seungbae Yu <dbadoy4874@gmail.com>
Date:   Sat Sep 3 00:48:26 2022 +0900

    trie: fix typo in comment (ethereum#25667)

commit 38e002f
Author: Seungbae Yu <dbadoy4874@gmail.com>
Date:   Sat Sep 3 00:47:29 2022 +0900

    rpc: check that "version" is "2.0" in request objects (ethereum#25570)

    The JSON-RPC spec requires the "version" field to be exactly "2.0",
    so we should verify that. This change is not backwards-compatible with
    sloppy client implementations, but I decided to go ahead with it anyway
    because the failure will be caught via the returned error.

commit 90711ef
Author: protolambda <proto@protolambda.com>
Date:   Fri Sep 2 17:40:41 2022 +0200

    node, rpc: add JWT auth support in client (ethereum#24911)

    This adds a generic mechanism for 'dial options' in the RPC client,
    and also implements a specific dial option for the JWT authentication
    mechanism used by the engine API. Some real tests for the server-side
    authentication handling are also added.

    Co-authored-by: Joshua Gutow <jgutow@optimism.io>
    Co-authored-by: Felix Lange <fjl@twurst.com>

commit 7f2890a
Author: Martin Holst Swende <martin@swende.se>
Date:   Fri Sep 2 17:28:33 2022 +0200

    eth/fetcher: throttle peers which deliver many invalid transactions (ethereum#25573)

    Co-authored-by: Felix Lange <fjl@twurst.com>

cmd/devp2p/internal/ethtest/helpers.go

@@ -357,9 +357,13 @@ func (s *Suite) waitAnnounce(conn *Conn, blockAnnouncement *NewBlock) error {
 				return fmt.Errorf("wrong block hash in announcement: expected %v, got %v", blockAnnouncement.Block.Hash(), hashes[0].Hash)
 			}
 			return nil
+
+		// ignore tx announcements from previous tests
 		case *NewPooledTransactionHashes:
-			// ignore tx announcements from previous tests
 			continue
+		case *Transactions:
+			continue
+
 		default:
 			return fmt.Errorf("unexpected: %s", pretty.Sdump(msg))
 		}

cmd/devp2p/internal/ethtest/suite.go

@@ -544,9 +544,13 @@ func (s *Suite) TestNewPooledTxs(t *utesting.T) {
 				t.Fatalf("unexpected number of txs requested: wanted %d, got %d", len(hashes), len(msg.GetPooledTransactionsPacket))
 			}
 			return
+
 		// ignore propagated txs from previous tests
 		case *NewPooledTransactionHashes:
 			continue
+		case *Transactions:
+			continue
+
 		// ignore block announcements from previous tests
 		case *NewBlockHashes:
 			continue

cmd/devp2p/internal/ethtest/transaction.go

@@ -29,7 +29,7 @@ import (
 	"github.com/ethereum/go-ethereum/params"
 )
 
-//var faucetAddr = common.HexToAddress("0x71562b71999873DB5b286dF957af199Ec94617F7")
+// var faucetAddr = common.HexToAddress("0x71562b71999873DB5b286dF957af199Ec94617F7")
 var faucetKey, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
 
 func (s *Suite) sendSuccessfulTxs(t *utesting.T) error {
@@ -192,10 +192,10 @@ func sendMultipleSuccessfulTxs(t *utesting.T, s *Suite, txs []*types.Transaction
 	nonce = txs[len(txs)-1].Nonce()
 
 	// Wait for the transaction announcement(s) and make sure all sent txs are being propagated.
-	// all txs should be announced within 3 announcements.
+	// all txs should be announced within a couple announcements.
 	recvHashes := make([]common.Hash, 0)
 
-	for i := 0; i < 3; i++ {
+	for i := 0; i < 20; i++ {
 		switch msg := recvConn.readAndServe(s.chain, timeout).(type) {
 		case *Transactions:
 			for _, tx := range *msg {

eth/fetcher/tx_fetcher.go

@@ -262,54 +262,72 @@ func (f *TxFetcher) Notify(peer string, hashes []common.Hash) error {
 // direct request replies. The differentiation is important so the fetcher can
 // re-schedule missing transactions as soon as possible.
 func (f *TxFetcher) Enqueue(peer string, txs []*types.Transaction, direct bool) error {
-	// Keep track of all the propagated transactions
-	if direct {
-		txReplyInMeter.Mark(int64(len(txs)))
-	} else {
-		txBroadcastInMeter.Mark(int64(len(txs)))
+	var (
+		inMeter          = txReplyInMeter
+		knownMeter       = txReplyKnownMeter
+		underpricedMeter = txReplyUnderpricedMeter
+		otherRejectMeter = txReplyOtherRejectMeter
+	)
+	if !direct {
+		inMeter = txBroadcastInMeter
+		knownMeter = txBroadcastKnownMeter
+		underpricedMeter = txBroadcastUnderpricedMeter
+		otherRejectMeter = txBroadcastOtherRejectMeter
 	}
+	// Keep track of all the propagated transactions
+	inMeter.Mark(int64(len(txs)))
+
 	// Push all the transactions into the pool, tracking underpriced ones to avoid
 	// re-requesting them and dropping the peer in case of malicious transfers.
 	var (
-		added       = make([]common.Hash, 0, len(txs))
-		duplicate   int64
-		underpriced int64
-		otherreject int64
+		added = make([]common.Hash, 0, len(txs))
 	)
-	errs := f.addTxs(txs)
-	for i, err := range errs {
-		// Track the transaction hash if the price is too low for us.
-		// Avoid re-request this transaction when we receive another
-		// announcement.
-		if errors.Is(err, core.ErrUnderpriced) || errors.Is(err, core.ErrReplaceUnderpriced) {
-			for f.underpriced.Cardinality() >= maxTxUnderpricedSetSize {
-				f.underpriced.Pop()
-			}
-			f.underpriced.Add(txs[i].Hash())
+	// proceed in batches
+	for i := 0; i < len(txs); i += 128 {
+		end := i + 128
+		if end > len(txs) {
+			end = len(txs)
 		}
-		// Track a few interesting failure types
-		switch {
-		case err == nil: // Noop, but need to handle to not count these
+		var (
+			duplicate   int64
+			underpriced int64
+			otherreject int64
+		)
+		batch := txs[i:end]
+		for j, err := range f.addTxs(batch) {
+			// Track the transaction hash if the price is too low for us.
+			// Avoid re-request this transaction when we receive another
+			// announcement.
+			if errors.Is(err, core.ErrUnderpriced) || errors.Is(err, core.ErrReplaceUnderpriced) {
+				for f.underpriced.Cardinality() >= maxTxUnderpricedSetSize {
+					f.underpriced.Pop()
+				}
+				f.underpriced.Add(batch[j].Hash())
+			}
+			// Track a few interesting failure types
+			switch {
+			case err == nil: // Noop, but need to handle to not count these
 
-		case errors.Is(err, core.ErrAlreadyKnown):
-			duplicate++
+			case errors.Is(err, core.ErrAlreadyKnown):
+				duplicate++
 
-		case errors.Is(err, core.ErrUnderpriced) || errors.Is(err, core.ErrReplaceUnderpriced):
-			underpriced++
+			case errors.Is(err, core.ErrUnderpriced) || errors.Is(err, core.ErrReplaceUnderpriced):
+				underpriced++
 
-		default:
-			otherreject++
+			default:
+				otherreject++
+			}
+			added = append(added, batch[j].Hash())
+		}
+		knownMeter.Mark(duplicate)
+		underpricedMeter.Mark(underpriced)
+		otherRejectMeter.Mark(otherreject)
+
+		// If 'other reject' is >25% of the deliveries in any batch, sleep a bit.
+		if otherreject > 128/4 {
+			time.Sleep(200 * time.Millisecond)
+			log.Warn("Peer delivering stale transactions", "peer", peer, "rejected", otherreject)
 		}
-		added = append(added, txs[i].Hash())
-	}
-	if direct {
-		txReplyKnownMeter.Mark(duplicate)
-		txReplyUnderpricedMeter.Mark(underpriced)
-		txReplyOtherRejectMeter.Mark(otherreject)
-	} else {
-		txBroadcastKnownMeter.Mark(duplicate)
-		txBroadcastUnderpricedMeter.Mark(underpriced)
-		txBroadcastOtherRejectMeter.Mark(otherreject)
 	}
 	select {
 	case f.cleanup <- &txDelivery{origin: peer, hashes: added, direct: direct}:

node/config.go

@@ -201,7 +201,7 @@ type Config struct {
 	// AllowUnprotectedTxs allows non EIP-155 protected transactions to be send over RPC.
 	AllowUnprotectedTxs bool `toml:",omitempty"`
 
-	// JWTSecret is the hex-encoded jwt secret.
+	// JWTSecret is the path to the hex-encoded jwt secret.
 	JWTSecret string `toml:",omitempty"`
 }
 

node/jwt_auth.go

@@ -0,0 +1,45 @@
+// Copyright 2022 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package node
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/golang-jwt/jwt/v4"
+)
+
+// NewJWTAuth creates an rpc client authentication provider that uses JWT. The
+// secret MUST be 32 bytes (256 bits) as defined by the Engine-API authentication spec.
+//
+// See https://github.com/ethereum/execution-apis/blob/main/src/engine/authentication.md
+// for more details about this authentication scheme.
+func NewJWTAuth(jwtsecret [32]byte) rpc.HTTPAuth {
+	return func(h http.Header) error {
+		token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
+			"iat": &jwt.NumericDate{Time: time.Now()},
+		})
+		s, err := token.SignedString(jwtsecret[:])
+		if err != nil {
+			return fmt.Errorf("failed to create JWT token: %w", err)
+		}
+		h.Set("Authorization", "Bearer "+s)
+		return nil
+	}
+}

node/node.go

@@ -668,6 +668,19 @@ func (n *Node) WSEndpoint() string {
 	return "ws://" + n.ws.listenAddr() + n.ws.wsConfig.prefix
 }
 
+// HTTPAuthEndpoint returns the URL of the authenticated HTTP server.
+func (n *Node) HTTPAuthEndpoint() string {
+	return "http://" + n.httpAuth.listenAddr()
+}
+
+// WSAuthEndpoint returns the current authenticated JSON-RPC over WebSocket endpoint.
+func (n *Node) WSAuthEndpoint() string {
+	if n.httpAuth.wsAllowed() {
+		return "ws://" + n.httpAuth.listenAddr() + n.httpAuth.wsConfig.prefix
+	}
+	return "ws://" + n.wsAuth.listenAddr() + n.wsAuth.wsConfig.prefix
+}
+
 // EventMux retrieves the event multiplexer used by all the network services in
 // the current protocol stack.
 func (n *Node) EventMux() *event.TypeMux {