Skip to content

Commit e2f2111

Browse files
MatusKyselMatusKysel
authored
p2p/discover: fix ENR filtering (#2770)
1 parent 2ae712b commit e2f2111

6 files changed

Lines changed: 55 additions & 44 deletions

File tree

‎cmd/utils/flags.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -974,7 +974,7 @@ var (
974974
Aliases: []string{"discv5"},
975975
Usage: "Enables the V5 discovery mechanism",
976976
Category: flags.NetworkingCategory,
977-
Value: true,
977+
Value: false,
978978
}
979979
NetrestrictFlag = &cli.StringFlag{
980980
Name: "netrestrict",

‎p2p/discover/node.go‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
package discover
1818

1919
import (
20-
"net"
2120
"slices"
2221
"sort"
2322
"time"
@@ -51,11 +50,6 @@ func unwrapNodes(ns []*tableNode) []*enode.Node {
5150
return result
5251
}
5352

54-
//nolint:unused
55-
func (n *tableNode) addr() *net.UDPAddr {
56-
return &net.UDPAddr{IP: n.IP(), Port: n.UDP()}
57-
}
58-
5953
func (n *tableNode) String() string {
6054
return n.Node.String()
6155
}

‎p2p/discover/table.go‎

Lines changed: 41 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -112,9 +112,10 @@ type bucket struct {
112112
}
113113

114114
type addNodeOp struct {
115-
node *enode.Node
116-
isInbound bool
117-
forceSetLive bool // for tests
115+
node *enode.Node
116+
isInbound bool
117+
forceSetLive bool // for tests
118+
syncExecution bool // for tests
118119
}
119120

120121
type trackRequestOp struct {
@@ -320,7 +321,7 @@ func (tab *Table) len() (n int) {
320321
//
321322
// The caller must not hold tab.mutex.
322323
func (tab *Table) addFoundNode(n *enode.Node, forceSetLive bool) bool {
323-
op := addNodeOp{node: n, isInbound: false, forceSetLive: forceSetLive}
324+
op := addNodeOp{node: n, isInbound: false, forceSetLive: forceSetLive, syncExecution: true}
324325
select {
325326
case tab.addNodeCh <- op:
326327
return <-tab.addNodeHandled
@@ -337,9 +338,20 @@ func (tab *Table) addFoundNode(n *enode.Node, forceSetLive bool) bool {
337338
// repeatedly.
338339
//
339340
// The caller must not hold tab.mutex.
340-
func (tab *Table) addInboundNode(n *enode.Node) bool {
341+
func (tab *Table) addInboundNode(n *enode.Node) {
341342
op := addNodeOp{node: n, isInbound: true}
342343
select {
344+
case tab.addNodeCh <- op:
345+
return
346+
case <-tab.closeReq:
347+
return
348+
}
349+
}
350+
351+
// Only for testing purposes
352+
func (tab *Table) addInboundNodeSync(n *enode.Node) bool {
353+
op := addNodeOp{node: n, isInbound: true, syncExecution: true}
354+
select {
343355
case tab.addNodeCh <- op:
344356
return <-tab.addNodeHandled
345357
case <-tab.closeReq:
@@ -387,10 +399,16 @@ loop:
387399
tab.revalidation.handleResponse(tab, r)
388400

389401
case op := <-tab.addNodeCh:
390-
tab.mutex.Lock()
391-
ok := tab.handleAddNode(op)
392-
tab.mutex.Unlock()
393-
tab.addNodeHandled <- ok
402+
// only happens in tests
403+
if op.syncExecution {
404+
ok := tab.handleAddNode(op)
405+
tab.addNodeHandled <- ok
406+
} else {
407+
// async execution as handleAddNode is blocking
408+
go func() {
409+
tab.handleAddNode(op)
410+
}()
411+
}
394412

395413
case op := <-tab.trackRequestCh:
396414
tab.handleTrackRequest(op)
@@ -468,9 +486,7 @@ func (tab *Table) loadSeedNodes() {
468486
addr, _ := seed.UDPEndpoint()
469487
tab.log.Trace("Found seed node in database", "id", seed.ID(), "addr", addr, "age", age)
470488
}
471-
tab.mutex.Lock()
472-
tab.handleAddNode(addNodeOp{node: seed, isInbound: false})
473-
tab.mutex.Unlock()
489+
go tab.handleAddNode(addNodeOp{node: seed, isInbound: false})
474490
}
475491
}
476492

@@ -492,16 +508,15 @@ func (tab *Table) bucketAtDistance(d int) *bucket {
492508
return tab.buckets[d-bucketMinDistance-1]
493509
}
494510

495-
//nolint:unused
496-
func (tab *Table) filterNode(n *tableNode) bool {
511+
func (tab *Table) filterNode(n *enode.Node) bool {
497512
if tab.enrFilter == nil {
498513
return false
499514
}
500-
if node, err := tab.net.RequestENR(n.Node); err != nil {
501-
tab.log.Debug("ENR request failed", "id", n.ID(), "addr", n.addr(), "err", err)
515+
if node, err := tab.net.RequestENR(n); err != nil {
516+
tab.log.Debug("ENR request failed", "id", n.ID(), "ipAddr", n.IPAddr(), "updPort", n.UDP(), "err", err)
502517
return false
503518
} else if !tab.enrFilter(node.Record()) {
504-
tab.log.Trace("ENR record filter out", "id", n.ID(), "addr", n.addr())
519+
tab.log.Trace("ENR record filter out", "id", n.ID(), "ipAddr", n.IPAddr(), "updPort", n.UDP())
505520
return true
506521
}
507522
return false
@@ -541,6 +556,13 @@ func (tab *Table) handleAddNode(req addNodeOp) bool {
541556
return false
542557
}
543558

559+
if tab.filterNode(req.node) {
560+
return false
561+
}
562+
563+
tab.mutex.Lock()
564+
defer tab.mutex.Unlock()
565+
544566
// For nodes from inbound contact, there is an additional safety measure: if the table
545567
// is still initializing the node is not added.
546568
if req.isInbound && !tab.isInitDone() {
@@ -570,11 +592,6 @@ func (tab *Table) handleAddNode(req addNodeOp) bool {
570592
wn.isValidatedLive = true
571593
}
572594

573-
// TODO(Matus): fix the filterNode feature
574-
// if tab.filterNode(wn) {
575-
// return false
576-
// }
577-
578595
b.entries = append(b.entries, wn)
579596
b.replacements = deleteNode(b.replacements, wn.ID())
580597
tab.nodeAdded(b, wn)
@@ -705,8 +722,6 @@ func (tab *Table) handleTrackRequest(op trackRequestOp) {
705722
}
706723

707724
tab.mutex.Lock()
708-
defer tab.mutex.Unlock()
709-
710725
b := tab.bucket(op.node.ID())
711726
// Remove the node from the local table if it fails to return anything useful too
712727
// many times, but only if there are enough other nodes in the bucket. This latter
@@ -715,10 +730,11 @@ func (tab *Table) handleTrackRequest(op trackRequestOp) {
715730
if fails >= maxFindnodeFailures && len(b.entries) >= bucketSize/4 {
716731
tab.deleteInBucket(b, op.node.ID())
717732
}
733+
tab.mutex.Unlock()
718734

719735
// Add found nodes.
720736
for _, n := range op.foundNodes {
721-
tab.handleAddNode(addNodeOp{n, false, false})
737+
go tab.handleAddNode(addNodeOp{n, false, false, false})
722738
}
723739
}
724740

‎p2p/discover/table_reval.go‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
package discover
1818

1919
import (
20-
"errors"
2120
"fmt"
2221
"math"
2322
"slices"
@@ -121,13 +120,7 @@ func (tab *Table) doRevalidate(resp revalidationResponse, node *enode.Node) {
121120
if err != nil {
122121
tab.log.Debug("ENR request failed", "id", node.ID(), "err", err)
123122
} else {
124-
if tab.enrFilter != nil && !tab.enrFilter(newrec.Record()) {
125-
tab.log.Trace("ENR record filter out", "id", node.ID(), "err", errors.New("filtered node"))
126-
// TODO: use didRespond to express failure temporarily
127-
resp.didRespond = false
128-
} else {
129-
resp.newRecord = newrec
130-
}
123+
resp.newRecord = newrec
131124
}
132125
}
133126

@@ -181,6 +174,11 @@ func (tr *tableRevalidation) handleResponse(tab *Table, resp revalidationRespons
181174
tab.log.Debug("Node revalidated", "b", b.index, "id", n.ID(), "checks", n.livenessChecks, "q", n.revalList.name)
182175
var endpointChanged bool
183176
if resp.newRecord != nil {
177+
if tab.enrFilter != nil && !tab.enrFilter(resp.newRecord.Record()) {
178+
tab.log.Trace("ENR record filter out", "id", n.ID())
179+
tab.deleteInBucket(b, n.ID())
180+
return
181+
}
184182
_, endpointChanged = tab.bumpInBucket(b, resp.newRecord, false)
185183
}
186184

‎p2p/discover/table_test.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -296,7 +296,7 @@ func TestTable_addInboundNode(t *testing.T) {
296296
newrec := n2.Record()
297297
newrec.Set(enr.IP{99, 99, 99, 99})
298298
n2v2 := enode.SignNull(newrec, n2.ID())
299-
tab.addInboundNode(n2v2)
299+
tab.addInboundNodeSync(n2v2)
300300
checkBucketContent(t, tab, []*enode.Node{n1, n2v2})
301301

302302
// Try updating n2 without sequence number change. The update is accepted
@@ -305,7 +305,7 @@ func TestTable_addInboundNode(t *testing.T) {
305305
newrec.Set(enr.IP{100, 100, 100, 100})
306306
newrec.SetSeq(n2.Seq())
307307
n2v3 := enode.SignNull(newrec, n2.ID())
308-
tab.addInboundNode(n2v3)
308+
tab.addInboundNodeSync(n2v3)
309309
checkBucketContent(t, tab, []*enode.Node{n1, n2v3})
310310
}
311311

@@ -349,13 +349,13 @@ func TestTable_addInboundNodeUpdateV4Accept(t *testing.T) {
349349
// Add a v4 node.
350350
key, _ := crypto.HexToECDSA("dd3757a8075e88d0f2b1431e7d3c5b1562e1c0aab9643707e8cbfcc8dae5cfe3")
351351
n1 := enode.NewV4(&key.PublicKey, net.IP{88, 77, 66, 1}, 9000, 9000)
352-
tab.addInboundNode(n1)
352+
tab.addInboundNodeSync(n1)
353353
checkBucketContent(t, tab, []*enode.Node{n1})
354354

355355
// Add an updated version with changed IP.
356356
// The update will be accepted because it is inbound.
357357
n1v2 := enode.NewV4(&key.PublicKey, net.IP{99, 99, 99, 99}, 9000, 9000)
358-
tab.addInboundNode(n1v2)
358+
tab.addInboundNodeSync(n1v2)
359359
checkBucketContent(t, tab, []*enode.Node{n1v2})
360360
}
361361

‎p2p/discover/v5_udp.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -437,6 +437,9 @@ func (t *UDPv5) verifyResponseNode(c *callV5, r *enr.Record, distances []uint, s
437437
if node.UDP() <= 1024 {
438438
return nil, errLowPort
439439
}
440+
if t.tab.enrFilter != nil && !t.tab.enrFilter(r) {
441+
return nil, errors.New("filtered by ENR filter")
442+
}
440443
if distances != nil {
441444
nd := enode.LogDist(c.id, node.ID())
442445
if !slices.Contains(distances, uint(nd)) {

0 commit comments

Comments
 (0)