p2p/discover: add initial discovery v5 implementation (#20750)

This adds an implementation of the current discovery v5 spec.

There is full integration with cmd/devp2p and enode.Iterator in this
version. In theory we could enable the new protocol as a replacement of
discovery v4 at any time. In practice, there will likely be a few more
changes to the spec and implementation before this can happen.

cmd/devp2p/crawl.go

@@ -20,14 +20,13 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 )
 
 type crawler struct {
 	input     nodeSet
 	output    nodeSet
-	disc      *discover.UDPv4
+	disc      resolver
 	iters     []enode.Iterator
 	inputIter enode.Iterator
 	ch        chan *enode.Node
@@ -37,7 +36,11 @@ type crawler struct {
 	revalidateInterval time.Duration
 }
 
-func newCrawler(input nodeSet, disc *discover.UDPv4, iters ...enode.Iterator) *crawler {
+type resolver interface {
+	RequestENR(*enode.Node) (*enode.Node, error)
+}
+
+func newCrawler(input nodeSet, disc resolver, iters ...enode.Iterator) *crawler {
 	c := &crawler{
 		input:     input,
 		output:    make(nodeSet, len(input)),

cmd/devp2p/discv4cmd.go

@@ -81,6 +81,18 @@ var (
 		Name:  "bootnodes",
 		Usage: "Comma separated nodes used for bootstrapping",
 	}
+	nodekeyFlag = cli.StringFlag{
+		Name:  "nodekey",
+		Usage: "Hex-encoded node key",
+	}
+	nodedbFlag = cli.StringFlag{
+		Name:  "nodedb",
+		Usage: "Nodes database location",
+	}
+	listenAddrFlag = cli.StringFlag{
+		Name:  "addr",
+		Usage: "Listening address",
+	}
 	crawlTimeoutFlag = cli.DurationFlag{
 		Name:  "timeout",
 		Usage: "Time limit for the crawl.",
@@ -172,28 +184,30 @@ func discv4Crawl(ctx *cli.Context) error {
 	return nil
 }
 
-func parseBootnodes(ctx *cli.Context) ([]*enode.Node, error) {
-	s := params.RinkebyBootnodes
-	if ctx.IsSet(bootnodesFlag.Name) {
-		s = strings.Split(ctx.String(bootnodesFlag.Name), ",")
-	}
-	nodes := make([]*enode.Node, len(s))
-	var err error
-	for i, record := range s {
-		nodes[i], err = parseNode(record)
-		if err != nil {
-			return nil, fmt.Errorf("invalid bootstrap node: %v", err)
-		}
-	}
-	return nodes, nil
-}
-
 // startV4 starts an ephemeral discovery V4 node.
 func startV4(ctx *cli.Context) *discover.UDPv4 {
-	socket, ln, cfg, err := listen()
+	ln, config := makeDiscoveryConfig(ctx)
+	socket := listen(ln, ctx.String(listenAddrFlag.Name))
+	disc, err := discover.ListenV4(socket, ln, config)
 	if err != nil {
 		exit(err)
 	}
+	return disc
+}
+
+func makeDiscoveryConfig(ctx *cli.Context) (*enode.LocalNode, discover.Config) {
+	var cfg discover.Config
+
+	if ctx.IsSet(nodekeyFlag.Name) {
+		key, err := crypto.HexToECDSA(ctx.String(nodekeyFlag.Name))
+		if err != nil {
+			exit(fmt.Errorf("-%s: %v", nodekeyFlag.Name, err))
+		}
+		cfg.PrivateKey = key
+	} else {
+		cfg.PrivateKey, _ = crypto.GenerateKey()
+	}
+
 	if commandHasFlag(ctx, bootnodesFlag) {
 		bn, err := parseBootnodes(ctx)
 		if err != nil {
@@ -201,26 +215,43 @@ func startV4(ctx *cli.Context) *discover.UDPv4 {
 		}
 		cfg.Bootnodes = bn
 	}
-	disc, err := discover.ListenV4(socket, ln, cfg)
+
+	dbpath := ctx.String(nodedbFlag.Name)
+	db, err := enode.OpenDB(dbpath)
 	if err != nil {
 		exit(err)
 	}
-	return disc
-}
-
-func listen() (*net.UDPConn, *enode.LocalNode, discover.Config, error) {
-	var cfg discover.Config
-	cfg.PrivateKey, _ = crypto.GenerateKey()
-	db, _ := enode.OpenDB("")
 	ln := enode.NewLocalNode(db, cfg.PrivateKey)
+	return ln, cfg
+}
 
-	socket, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IP{0, 0, 0, 0}})
+func listen(ln *enode.LocalNode, addr string) *net.UDPConn {
+	if addr == "" {
+		addr = "0.0.0.0:0"
+	}
+	socket, err := net.ListenPacket("udp4", addr)
 	if err != nil {
-		db.Close()
-		return nil, nil, cfg, err
+		exit(err)
 	}
-	addr := socket.LocalAddr().(*net.UDPAddr)
+	usocket := socket.(*net.UDPConn)
+	uaddr := socket.LocalAddr().(*net.UDPAddr)
 	ln.SetFallbackIP(net.IP{127, 0, 0, 1})
-	ln.SetFallbackUDP(addr.Port)
-	return socket, ln, cfg, nil
+	ln.SetFallbackUDP(uaddr.Port)
+	return usocket
+}
+
+func parseBootnodes(ctx *cli.Context) ([]*enode.Node, error) {
+	s := params.RinkebyBootnodes
+	if ctx.IsSet(bootnodesFlag.Name) {
+		s = strings.Split(ctx.String(bootnodesFlag.Name), ",")
+	}
+	nodes := make([]*enode.Node, len(s))
+	var err error
+	for i, record := range s {
+		nodes[i], err = parseNode(record)
+		if err != nil {
+			return nil, fmt.Errorf("invalid bootstrap node: %v", err)
+		}
+	}
+	return nodes, nil
 }

cmd/devp2p/discv5cmd.go

+// Copyright 2019 The go-ethereum Authors
+// This file is part of go-ethereum.
+//
+// go-ethereum is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// go-ethereum 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with go-ethereum. If not, see <http://www.gnu.org/licenses/>.
+
+package main
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"gopkg.in/urfave/cli.v1"
+)
+
+var (
+	discv5Command = cli.Command{
+		Name:  "discv5",
+		Usage: "Node Discovery v5 tools",
+		Subcommands: []cli.Command{
+			discv5PingCommand,
+			discv5ResolveCommand,
+			discv5CrawlCommand,
+			discv5ListenCommand,
+		},
+	}
+	discv5PingCommand = cli.Command{
+		Name:   "ping",
+		Usage:  "Sends ping to a node",
+		Action: discv5Ping,
+	}
+	discv5ResolveCommand = cli.Command{
+		Name:   "resolve",
+		Usage:  "Finds a node in the DHT",
+		Action: discv5Resolve,
+		Flags:  []cli.Flag{bootnodesFlag},
+	}
+	discv5CrawlCommand = cli.Command{
+		Name:   "crawl",
+		Usage:  "Updates a nodes.json file with random nodes found in the DHT",
+		Action: discv5Crawl,
+		Flags:  []cli.Flag{bootnodesFlag, crawlTimeoutFlag},
+	}
+	discv5ListenCommand = cli.Command{
+		Name:   "listen",
+		Usage:  "Runs a node",
+		Action: discv5Listen,
+		Flags: []cli.Flag{
+			bootnodesFlag,
+			nodekeyFlag,
+			nodedbFlag,
+			listenAddrFlag,
+		},
+	}
+)
+
+func discv5Ping(ctx *cli.Context) error {
+	n := getNodeArg(ctx)
+	disc := startV5(ctx)
+	defer disc.Close()
+
+	fmt.Println(disc.Ping(n))
+	return nil
+}
+
+func discv5Resolve(ctx *cli.Context) error {
+	n := getNodeArg(ctx)
+	disc := startV5(ctx)
+	defer disc.Close()
+
+	fmt.Println(disc.Resolve(n))
+	return nil
+}
+
+func discv5Crawl(ctx *cli.Context) error {
+	if ctx.NArg() < 1 {
+		return fmt.Errorf("need nodes file as argument")
+	}
+	nodesFile := ctx.Args().First()
+	var inputSet nodeSet
+	if common.FileExist(nodesFile) {
+		inputSet = loadNodesJSON(nodesFile)
+	}
+
+	disc := startV5(ctx)
+	defer disc.Close()
+	c := newCrawler(inputSet, disc, disc.RandomNodes())
+	c.revalidateInterval = 10 * time.Minute
+	output := c.run(ctx.Duration(crawlTimeoutFlag.Name))
+	writeNodesJSON(nodesFile, output)
+	return nil
+}
+
+func discv5Listen(ctx *cli.Context) error {
+	disc := startV5(ctx)
+	defer disc.Close()
+
+	fmt.Println(disc.Self())
+	select {}
+}
+
+// startV5 starts an ephemeral discovery v5 node.
+func startV5(ctx *cli.Context) *discover.UDPv5 {
+	ln, config := makeDiscoveryConfig(ctx)
+	socket := listen(ln, ctx.String(listenAddrFlag.Name))
+	disc, err := discover.ListenV5(socket, ln, config)
+	if err != nil {
+		exit(err)
+	}
+	return disc
+}

cmd/devp2p/main.go

@@ -59,6 +59,7 @@ func init() {
 	app.Commands = []cli.Command{
 		enrdumpCommand,
 		discv4Command,
+		discv5Command,
 		dnsCommand,
 		nodesetCommand,
 	}

p2p/discover/common.go

@@ -20,8 +20,10 @@ import (
 	"crypto/ecdsa"
 	"net"
 
+	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/p2p/enr"
 	"github.com/ethereum/go-ethereum/p2p/netutil"
 )
 
@@ -39,10 +41,25 @@ type Config struct {
 	PrivateKey *ecdsa.PrivateKey
 
 	// These settings are optional:
-	NetRestrict *netutil.Netlist  // network whitelist
-	Bootnodes   []*enode.Node     // list of bootstrap nodes
-	Unhandled   chan<- ReadPacket // unhandled packets are sent on this channel
-	Log         log.Logger        // if set, log messages go here
+	NetRestrict  *netutil.Netlist   // network whitelist
+	Bootnodes    []*enode.Node      // list of bootstrap nodes
+	Unhandled    chan<- ReadPacket  // unhandled packets are sent on this channel
+	Log          log.Logger         // if set, log messages go here
+	ValidSchemes enr.IdentityScheme // allowed identity schemes
+	Clock        mclock.Clock
+}
+
+func (cfg Config) withDefaults() Config {
+	if cfg.Log == nil {
+		cfg.Log = log.Root()
+	}
+	if cfg.ValidSchemes == nil {
+		cfg.ValidSchemes = enode.ValidSchemes
+	}
+	if cfg.Clock == nil {
+		cfg.Clock = mclock.System{}
+	}
+	return cfg
 }
 
 // ListenUDP starts listening for discovery packets on the given UDP socket.
@@ -51,8 +68,15 @@ func ListenUDP(c UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv4, error) {
 }
 
 // ReadPacket is a packet that couldn't be handled. Those packets are sent to the unhandled
-// channel if configured. This is exported for internal use, do not use this type.
+// channel if configured.
 type ReadPacket struct {
 	Data []byte
 	Addr *net.UDPAddr
 }
+
+func min(x, y int) int {
+	if x > y {
+		return y
+	}
+	return x
+}

p2p/discover/lookup.go

@@ -150,7 +150,7 @@ func (it *lookup) query(n *node, reply chan<- []*node) {
 	} else if len(r) == 0 {
 		fails++
 		it.tab.db.UpdateFindFails(n.ID(), n.IP(), fails)
-		it.tab.log.Trace("Findnode failed", "id", n.ID(), "failcount", fails, "err", err)
+		it.tab.log.Trace("Findnode failed", "id", n.ID(), "failcount", fails, "results", len(r), "err", err)
 		if fails >= maxFindnodeFailures {
 			it.tab.log.Trace("Too many findnode failures, dropping", "id", n.ID(), "failcount", fails)
 			it.tab.delete(n)

p2p/discover/node.go

@@ -18,6 +18,7 @@ package discover
 
 import (
 	"crypto/ecdsa"
+	"crypto/elliptic"
 	"errors"
 	"math/big"
 	"net"
@@ -45,13 +46,13 @@ func encodePubkey(key *ecdsa.PublicKey) encPubkey {
 	return e
 }
 
-func decodePubkey(e encPubkey) (*ecdsa.PublicKey, error) {
-	p := &ecdsa.PublicKey{Curve: crypto.S256(), X: new(big.Int), Y: new(big.Int)}
+func decodePubkey(curve elliptic.Curve, e encPubkey) (*ecdsa.PublicKey, error) {
+	p := &ecdsa.PublicKey{Curve: curve, X: new(big.Int), Y: new(big.Int)}
 	half := len(e) / 2
 	p.X.SetBytes(e[:half])
 	p.Y.SetBytes(e[half:])
 	if !p.Curve.IsOnCurve(p.X, p.Y) {
-		return nil, errors.New("invalid secp256k1 curve point")
+		return nil, errors.New("invalid curve point")
 	}
 	return p, nil
 }

p2p/discover/table.go

@@ -424,6 +424,10 @@ func (tab *Table) len() (n int) {
 // bucket returns the bucket for the given node ID hash.
 func (tab *Table) bucket(id enode.ID) *bucket {
 	d := enode.LogDist(tab.self().ID(), id)
+	return tab.bucketAtDistance(d)
+}
+
+func (tab *Table) bucketAtDistance(d int) *bucket {
 	if d <= bucketMinDistance {
 		return tab.buckets[0]
 	}

p2p/discover/table_util_test.go

@@ -24,7 +24,6 @@ import (
 	"fmt"
 	"math/rand"
 	"net"
-	"reflect"
 	"sort"
 	"sync"
 
@@ -56,6 +55,23 @@ func nodeAtDistance(base enode.ID, ld int, ip net.IP) *node {
 	return wrapNode(enode.SignNull(&r, idAtDistance(base, ld)))
 }
 
+// nodesAtDistance creates n nodes for which enode.LogDist(base, node.ID()) == ld.
+func nodesAtDistance(base enode.ID, ld int, n int) []*enode.Node {
+	results := make([]*enode.Node, n)
+	for i := range results {
+		results[i] = unwrapNode(nodeAtDistance(base, ld, intIP(i)))
+	}
+	return results
+}
+
+func nodesToRecords(nodes []*enode.Node) []*enr.Record {
+	records := make([]*enr.Record, len(nodes))
+	for i := range nodes {
+		records[i] = nodes[i].Record()
+	}
+	return records
+}
+
 // idAtDistance returns a random hash such that enode.LogDist(a, b) == n
 func idAtDistance(a enode.ID, n int) (b enode.ID) {
 	if n == 0 {
@@ -173,9 +189,16 @@ func hasDuplicates(slice []*node) bool {
 }
 
 func checkNodesEqual(got, want []*enode.Node) error {
-	if reflect.DeepEqual(got, want) {
-		return nil
+	if len(got) == len(want) {
+		for i := range got {
+			if !nodeEqual(got[i], want[i]) {
+				goto NotEqual
+			}
+			return nil
+		}
 	}
+
+NotEqual:
 	output := new(bytes.Buffer)
 	fmt.Fprintf(output, "got %d nodes:\n", len(got))
 	for _, n := range got {
@@ -188,6 +211,10 @@ func checkNodesEqual(got, want []*enode.Node) error {
 	return errors.New(output.String())
 }
 
+func nodeEqual(n1 *enode.Node, n2 *enode.Node) bool {
+	return n1.ID() == n2.ID() && n1.IP().Equal(n2.IP())
+}
+
 func sortByID(nodes []*enode.Node) {
 	sort.Slice(nodes, func(i, j int) bool {
 		return string(nodes[i].ID().Bytes()) < string(nodes[j].ID().Bytes())

p2p/discover/v4_lookup_test.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/p2p/enr"
 )
 
 func TestUDPv4_Lookup(t *testing.T) {
@@ -32,7 +33,7 @@ func TestUDPv4_Lookup(t *testing.T) {
 	test := newUDPTest(t)
 
 	// Lookup on empty table returns no nodes.
-	targetKey, _ := decodePubkey(lookupTestnet.target)
+	targetKey, _ := decodePubkey(crypto.S256(), lookupTestnet.target)
 	if results := test.udp.LookupPubkey(targetKey); len(results) > 0 {
 		t.Fatalf("lookup on empty table returned %d results: %#v", len(results), results)
 	}
@@ -59,15 +60,7 @@ func TestUDPv4_Lookup(t *testing.T) {
 	if len(results) != bucketSize {
 		t.Errorf("wrong number of results: got %d, want %d", len(results), bucketSize)
 	}
-	if hasDuplicates(wrapNodes(results)) {
-		t.Errorf("result set contains duplicate entries")
-	}
-	if !sortedByDistanceTo(lookupTestnet.target.id(), wrapNodes(results)) {
-		t.Errorf("result set not sorted by distance to target")
-	}
-	if err := checkNodesEqual(results, lookupTestnet.closest(bucketSize)); err != nil {
-		t.Errorf("results aren't the closest %d nodes\n%v", bucketSize, err)
-	}
+	checkLookupResults(t, lookupTestnet, results)
 }
 
 func TestUDPv4_LookupIterator(t *testing.T) {
@@ -156,6 +149,26 @@ func serveTestnet(test *udpTest, testnet *preminedTestnet) {
 	}
 }
 
+// checkLookupResults verifies that the results of a lookup are the closest nodes to
+// the testnet's target.
+func checkLookupResults(t *testing.T, tn *preminedTestnet, results []*enode.Node) {
+	t.Helper()
+	t.Logf("results:")
+	for _, e := range results {
+		t.Logf("  ld=%d, %x", enode.LogDist(tn.target.id(), e.ID()), e.ID().Bytes())
+	}
+	if hasDuplicates(wrapNodes(results)) {
+		t.Errorf("result set contains duplicate entries")
+	}
+	if !sortedByDistanceTo(tn.target.id(), wrapNodes(results)) {
+		t.Errorf("result set not sorted by distance to target")
+	}
+	wantNodes := tn.closest(len(results))
+	if err := checkNodesEqual(results, wantNodes); err != nil {
+		t.Error(err)
+	}
+}
+
 // This is the test network for the Lookup test.
 // The nodes were obtained by running lookupTestnet.mine with a random NodeID as target.
 var lookupTestnet = &preminedTestnet{
@@ -242,8 +255,12 @@ func (tn *preminedTestnet) nodes() []*enode.Node {
 
 func (tn *preminedTestnet) node(dist, index int) *enode.Node {
 	key := tn.dists[dist][index]
-	ip := net.IP{127, byte(dist >> 8), byte(dist), byte(index)}
-	return enode.NewV4(&key.PublicKey, ip, 0, 5000)
+	rec := new(enr.Record)
+	rec.Set(enr.IP{127, byte(dist >> 8), byte(dist), byte(index)})
+	rec.Set(enr.UDP(5000))
+	enode.SignV4(rec, key)
+	n, _ := enode.New(enode.ValidSchemes, rec)
+	return n
 }
 
 func (tn *preminedTestnet) nodeByAddr(addr *net.UDPAddr) (*enode.Node, *ecdsa.PrivateKey) {
@@ -261,6 +278,19 @@ func (tn *preminedTestnet) nodesAtDistance(dist int) []rpcNode {
 	return result
 }
 
+func (tn *preminedTestnet) neighborsAtDistance(base *enode.Node, distance uint, elems int) []*enode.Node {
+	nodes := nodesByDistance{target: base.ID()}
+	for d := range lookupTestnet.dists {
+		for i := range lookupTestnet.dists[d] {
+			n := lookupTestnet.node(d, i)
+			if uint(enode.LogDist(n.ID(), base.ID())) == distance {
+				nodes.push(wrapNode(n), elems)
+			}
+		}
+	}
+	return unwrapNodes(nodes.entries)
+}
+
 func (tn *preminedTestnet) closest(n int) (nodes []*enode.Node) {
 	for d := range tn.dists {
 		for i := range tn.dists[d] {

p2p/discover/v4_udp.go

@@ -47,6 +47,7 @@ var (
 	errTimeout          = errors.New("RPC timeout")
 	errClockWarp        = errors.New("reply deadline too far in the future")
 	errClosed           = errors.New("socket closed")
+	errLowPort          = errors.New("low port")
 )
 
 const (
@@ -176,7 +177,7 @@ func (t *UDPv4) nodeFromRPC(sender *net.UDPAddr, rn rpcNode) (*node, error) {
 	if t.netrestrict != nil && !t.netrestrict.Contains(rn.IP) {
 		return nil, errors.New("not contained in netrestrict whitelist")
 	}
-	key, err := decodePubkey(rn.ID)
+	key, err := decodePubkey(crypto.S256(), rn.ID)
 	if err != nil {
 		return nil, err
 	}
@@ -209,7 +210,7 @@ type UDPv4 struct {
 	addReplyMatcher chan *replyMatcher
 	gotreply        chan reply
 	closeCtx        context.Context
-	cancelCloseCtx  func()
+	cancelCloseCtx  context.CancelFunc
 }
 
 // replyMatcher represents a pending reply.
@@ -258,6 +259,7 @@ type reply struct {
 }
 
 func ListenV4(c UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv4, error) {
+	cfg = cfg.withDefaults()
 	closeCtx, cancel := context.WithCancel(context.Background())
 	t := &UDPv4{
 		conn:            c,
@@ -271,9 +273,6 @@ func ListenV4(c UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv4, error) {
 		cancelCloseCtx:  cancel,
 		log:             cfg.Log,
 	}
-	if t.log == nil {
-		t.log = log.Root()
-	}
 
 	tab, err := newTable(t, ln.Database(), cfg.Bootnodes, t.log)
 	if err != nil {
@@ -812,7 +811,7 @@ func (req *pingV4) preverify(t *UDPv4, from *net.UDPAddr, fromID enode.ID, fromK
 	if expired(req.Expiration) {
 		return errExpired
 	}
-	key, err := decodePubkey(fromKey)
+	key, err := decodePubkey(crypto.S256(), fromKey)
 	if err != nil {
 		return errors.New("invalid public key")
 	}

p2p/discover/v4_udp_test.go

@@ -41,10 +41,6 @@ import (
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
-func init() {
-	spew.Config.DisableMethods = true
-}
-
 // shared test variables
 var (
 	futureExp          = uint64(time.Now().Add(10 * time.Hour).Unix())
@@ -117,9 +113,12 @@ func (test *udpTest) packetInFrom(wantError error, key *ecdsa.PrivateKey, addr *
 func (test *udpTest) waitPacketOut(validate interface{}) (closed bool) {
 	test.t.Helper()
 
-	dgram, ok := test.pipe.receive()
-	if !ok {
+	dgram, err := test.pipe.receive()
+	if err == errClosed {
 		return true
+	} else if err != nil {
+		test.t.Error("packet receive error:", err)
+		return false
 	}
 	p, _, hash, err := decodeV4(dgram.data)
 	if err != nil {
@@ -671,17 +670,30 @@ func (c *dgramPipe) LocalAddr() net.Addr {
 	return &net.UDPAddr{IP: testLocal.IP, Port: int(testLocal.UDP)}
 }
 
-func (c *dgramPipe) receive() (dgram, bool) {
+func (c *dgramPipe) receive() (dgram, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	for len(c.queue) == 0 && !c.closed {
+
+	var timedOut bool
+	timer := time.AfterFunc(3*time.Second, func() {
+		c.mu.Lock()
+		timedOut = true
+		c.mu.Unlock()
+		c.cond.Broadcast()
+	})
+	defer timer.Stop()
+
+	for len(c.queue) == 0 && !c.closed && !timedOut {
 		c.cond.Wait()
 	}
 	if c.closed {
-		return dgram{}, false
+		return dgram{}, errClosed
+	}
+	if timedOut {
+		return dgram{}, errTimeout
 	}
 	p := c.queue[0]
 	copy(c.queue, c.queue[1:])
 	c.queue = c.queue[:len(c.queue)-1]
-	return p, true
+	return p, nil
 }

p2p/discover/v5_session.go

+// Copyright 2020 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 discover
+
+import (
+	crand "crypto/rand"
+
+	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/hashicorp/golang-lru/simplelru"
+)
+
+// The sessionCache keeps negotiated encryption keys and
+// state for in-progress handshakes in the Discovery v5 wire protocol.
+type sessionCache struct {
+	sessions   *simplelru.LRU
+	handshakes map[sessionID]*whoareyouV5
+	clock      mclock.Clock
+}
+
+// sessionID identifies a session or handshake.
+type sessionID struct {
+	id   enode.ID
+	addr string
+}
+
+// session contains session information
+type session struct {
+	writeKey     []byte
+	readKey      []byte
+	nonceCounter uint32
+}
+
+func newSessionCache(maxItems int, clock mclock.Clock) *sessionCache {
+	cache, err := simplelru.NewLRU(maxItems, nil)
+	if err != nil {
+		panic("can't create session cache")
+	}
+	return &sessionCache{
+		sessions:   cache,
+		handshakes: make(map[sessionID]*whoareyouV5),
+		clock:      clock,
+	}
+}
+
+// nextNonce creates a nonce for encrypting a message to the given session.
+func (sc *sessionCache) nextNonce(id enode.ID, addr string) []byte {
+	n := make([]byte, gcmNonceSize)
+	crand.Read(n)
+	return n
+}
+
+// session returns the current session for the given node, if any.
+func (sc *sessionCache) session(id enode.ID, addr string) *session {
+	item, ok := sc.sessions.Get(sessionID{id, addr})
+	if !ok {
+		return nil
+	}
+	return item.(*session)
+}
+
+// readKey returns the current read key for the given node.
+func (sc *sessionCache) readKey(id enode.ID, addr string) []byte {
+	if s := sc.session(id, addr); s != nil {
+		return s.readKey
+	}
+	return nil
+}
+
+// writeKey returns the current read key for the given node.
+func (sc *sessionCache) writeKey(id enode.ID, addr string) []byte {
+	if s := sc.session(id, addr); s != nil {
+		return s.writeKey
+	}
+	return nil
+}
+
+// storeNewSession stores new encryption keys in the cache.
+func (sc *sessionCache) storeNewSession(id enode.ID, addr string, r, w []byte) {
+	sc.sessions.Add(sessionID{id, addr}, &session{
+		readKey: r, writeKey: w,
+	})
+}
+
+// getHandshake gets the handshake challenge we previously sent to the given remote node.
+func (sc *sessionCache) getHandshake(id enode.ID, addr string) *whoareyouV5 {
+	return sc.handshakes[sessionID{id, addr}]
+}
+
+// storeSentHandshake stores the handshake challenge sent to the given remote node.
+func (sc *sessionCache) storeSentHandshake(id enode.ID, addr string, challenge *whoareyouV5) {
+	challenge.sent = sc.clock.Now()
+	sc.handshakes[sessionID{id, addr}] = challenge
+}
+
+// deleteHandshake deletes handshake data for the given node.
+func (sc *sessionCache) deleteHandshake(id enode.ID, addr string) {
+	delete(sc.handshakes, sessionID{id, addr})
+}
+
+// handshakeGC deletes timed-out handshakes.
+func (sc *sessionCache) handshakeGC() {
+	deadline := sc.clock.Now().Add(-handshakeTimeout)
+	for key, challenge := range sc.handshakes {
+		if challenge.sent < deadline {
+			delete(sc.handshakes, key)
+		}
+	}
+}

p2p/enode/nodedb.go

@@ -41,6 +41,7 @@ const (
 	dbNodePrefix   = "n:"      // Identifier to prefix node entries with
 	dbLocalPrefix  = "local:"
 	dbDiscoverRoot = "v4"
+	dbDiscv5Root   = "v5"
 
 	// These fields are stored per ID and IP, the full key is "n:<ID>:v4:<IP>:findfail".
 	// Use nodeItemKey to create those keys.
@@ -172,6 +173,16 @@ func splitNodeItemKey(key []byte) (id ID, ip net.IP, field string) {
 	return id, ip, field
 }
 
+func v5Key(id ID, ip net.IP, field string) []byte {
+	return bytes.Join([][]byte{
+		[]byte(dbNodePrefix),
+		id[:],
+		[]byte(dbDiscv5Root),
+		ip.To16(),
+		[]byte(field),
+	}, []byte{':'})
+}
+
 // localItemKey returns the key of a local node item.
 func localItemKey(id ID, field string) []byte {
 	key := append([]byte(dbLocalPrefix), id[:]...)
@@ -378,6 +389,16 @@ func (db *DB) UpdateFindFails(id ID, ip net.IP, fails int) error {
 	return db.storeInt64(nodeItemKey(id, ip, dbNodeFindFails), int64(fails))
 }
 
+// FindFailsV5 retrieves the discv5 findnode failure counter.
+func (db *DB) FindFailsV5(id ID, ip net.IP) int {
+	return int(db.fetchInt64(v5Key(id, ip, dbNodeFindFails)))
+}
+
+// UpdateFindFailsV5 stores the discv5 findnode failure counter.
+func (db *DB) UpdateFindFailsV5(id ID, ip net.IP, fails int) error {
+	return db.storeInt64(v5Key(id, ip, dbNodeFindFails), int64(fails))
+}
+
 // LocalSeq retrieves the local record sequence counter.
 func (db *DB) localSeq(id ID) uint64 {
 	return db.fetchUint64(localItemKey(id, dbLocalSeq))

p2p/enode/nodedb_test.go

@@ -462,3 +462,14 @@ func TestDBExpiration(t *testing.T) {
 		}
 	}
 }
+
+// This test checks that expiration works when discovery v5 data is present
+// in the database.
+func TestDBExpireV5(t *testing.T) {
+	db, _ := OpenDB("")
+	defer db.Close()
+
+	ip := net.IP{127, 0, 0, 1}
+	db.UpdateFindFailsV5(ID{}, ip, 4)
+	db.expireNodes()
+}