Merge pull request #10102 from yyforyongyu/fix-UpdatesInHorizon

Catch bad gossip peer and fix `UpdatesInHorizon`

Author: yyforyongyu

config.go

@@ -720,6 +720,7 @@ func DefaultConfig() Config {
 			MsgRateBytes:          discovery.DefaultMsgBytesPerSecond,
 			MsgBurstBytes:         discovery.DefaultMsgBytesBurst,
 			FilterConcurrency:     discovery.DefaultFilterConcurrency,
+			BanThreshold:          discovery.DefaultBanThreshold,
 		},
 		Invoices: &lncfg.Invoices{
 			HoldExpiryDelta: lncfg.DefaultHoldInvoiceExpiryDelta,

discovery/ban.go

@@ -2,6 +2,7 @@ package discovery
 
 import (
 	"errors"
+	"math"
 	"sync"
 	"time"
 
@@ -13,15 +14,14 @@ import (
 )
 
 const (
+	// DefaultBanThreshold is the default value to be used for banThreshold.
+	DefaultBanThreshold = 100
+
 	// maxBannedPeers limits the maximum number of banned pubkeys that
 	// we'll store.
 	// TODO(eugene): tune.
 	maxBannedPeers = 10_000
 
-	// banThreshold is the point at which non-channel peers will be banned.
-	// TODO(eugene): tune.
-	banThreshold = 100
-
 	// banTime is the amount of time that the non-channel peer will be
 	// banned for. Channel announcements from channel peers will be dropped
 	// if it's not one of our channels.
@@ -126,7 +126,7 @@ func (c *cachedBanInfo) Size() (uint64, error) {
 }
 
 // isBanned returns true if the ban score is greater than the ban threshold.
-func (c *cachedBanInfo) isBanned() bool {
+func (c *cachedBanInfo) isBanned(banThreshold uint64) bool {
 	return c.score >= banThreshold
 }
 
@@ -144,15 +144,26 @@ type banman struct {
 
 	wg   sync.WaitGroup
 	quit chan struct{}
+
+	// banThreshold is the point at which non-channel peers will be banned.
+	banThreshold uint64
 }
 
 // newBanman creates a new banman with the default maxBannedPeers.
-func newBanman() *banman {
+func newBanman(banThreshold uint64) *banman {
+	// If the ban threshold is set to 0, we'll use the max value to
+	// effectively disable banning.
+	if banThreshold == 0 {
+		log.Warn("Banning is disabled due to zero banThreshold")
+		banThreshold = math.MaxUint64
+	}
+
 	return &banman{
 		peerBanIndex: lru.NewCache[[33]byte, *cachedBanInfo](
 			maxBannedPeers,
 		),
-		quit: make(chan struct{}),
+		quit:         make(chan struct{}),
+		banThreshold: banThreshold,
 	}
 }
 
@@ -193,7 +204,7 @@ func (b *banman) purgeBanEntries() {
 	keysToRemove := make([][33]byte, 0)
 
 	sweepEntries := func(pubkey [33]byte, banInfo *cachedBanInfo) bool {
-		if banInfo.isBanned() {
+		if banInfo.isBanned(b.banThreshold) {
 			// If the peer is banned, check if the ban timer has
 			// expired.
 			if banInfo.lastUpdate.Add(banTime).Before(time.Now()) {
@@ -227,7 +238,7 @@ func (b *banman) isBanned(pubkey [33]byte) bool {
 		return false
 
 	default:
-		return banInfo.isBanned()
+		return banInfo.isBanned(b.banThreshold)
 	}
 }
 

discovery/ban_test.go

@@ -12,12 +12,14 @@ import (
 func TestPurgeBanEntries(t *testing.T) {
 	t.Parallel()
 
-	b := newBanman()
+	testBanThreshold := uint64(10)
+
+	b := newBanman(testBanThreshold)
 
 	// Ban a peer by repeatedly incrementing its ban score.
 	peer1 := [33]byte{0x00}
 
-	for i := 0; i < banThreshold; i++ {
+	for range testBanThreshold {
 		b.incrementBanScore(peer1)
 	}
 

discovery/chan_series.go

@@ -120,6 +120,10 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 	if err != nil {
 		return nil, err
 	}
+
+	// nodesFromChan records the nodes seen from the channels.
+	nodesFromChan := make(map[[33]byte]struct{}, len(chansInHorizon)*2)
+
 	for _, channel := range chansInHorizon {
 		// If the channel hasn't been fully advertised yet, or is a
 		// private channel, then we'll skip it as we can't construct a
@@ -136,7 +140,21 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 			return nil, err
 		}
 
-		updates = append(updates, chanAnn)
+		// Create a slice to hold the `channel_announcement` and
+		// potentially two `channel_update` msgs.
+		//
+		// NOTE: Based on BOLT7, if a channel_announcement has no
+		// corresponding channel_updates, we must not send the
+		// channel_announcement. Thus we use this slice to decide we
+		// want to send this `channel_announcement` or not. By the end
+		// of the operation, if the len of the slice is 1, we will not
+		// send the `channel_announcement`. Otherwise, when sending the
+		// msgs, the `channel_announcement` must be sent prior to any
+		// corresponding `channel_update` or `node_annoucement`, that's
+		// why we create a slice here to maintain the order.
+		chanUpdates := make([]lnwire.Message, 0, 3)
+		chanUpdates = append(chanUpdates, chanAnn)
+
 		if edge1 != nil {
 			// We don't want to send channel updates that don't
 			// conform to the spec (anymore).
@@ -145,18 +163,32 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 				log.Errorf("not sending invalid channel "+
 					"update %v: %v", edge1, err)
 			} else {
-				updates = append(updates, edge1)
+				chanUpdates = append(chanUpdates, edge1)
 			}
 		}
+
 		if edge2 != nil {
 			err := netann.ValidateChannelUpdateFields(0, edge2)
 			if err != nil {
 				log.Errorf("not sending invalid channel "+
 					"update %v: %v", edge2, err)
 			} else {
-				updates = append(updates, edge2)
+				chanUpdates = append(chanUpdates, edge2)
 			}
 		}
+
+		// If there's no corresponding `channel_update` to send, skip
+		// sending this `channel_announcement`.
+		if len(chanUpdates) < 2 {
+			continue
+		}
+
+		// Append the all the msgs to the slice.
+		updates = append(updates, chanUpdates...)
+
+		// Record the nodes seen.
+		nodesFromChan[channel.Info.NodeKey1Bytes] = struct{}{}
+		nodesFromChan[channel.Info.NodeKey2Bytes] = struct{}{}
 	}
 
 	// Next, we'll send out all the node announcements that have an update
@@ -168,8 +200,15 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 	if err != nil {
 		return nil, err
 	}
+
 	for _, nodeAnn := range nodeAnnsInHorizon {
-		nodeAnn := nodeAnn
+		// If this node has not been seen in the above channels, we can
+		// skip sending its NodeAnnouncement.
+		if _, seen := nodesFromChan[nodeAnn.PubKeyBytes]; !seen {
+			log.Debugf("Skipping forwarding as node %x not found "+
+				"in channel announcement", nodeAnn.PubKeyBytes)
+			continue
+		}
 
 		// Ensure we only forward nodes that are publicly advertised to
 		// prevent leaking information about nodes.