From 15f17633aa483b15374d716e9cec850c38f42c97 Mon Sep 17 00:00:00 2001 From: Eugene Siegel Date: Wed, 15 Jan 2025 13:10:48 -0500 Subject: [PATCH 1/7] channeldb: FetchPermAndTempPeers to load access perms on startup We introduce a new func FetchPermAndTempPeers that returns two maps. The first map indicates the nodes that will have "protected" access to the server. The second map indicates the nodes that have "temporary" access to the server. This will be used in a future commit in the server.go code. --- channeldb/channel_test.go | 30 ++++++ channeldb/db.go | 188 ++++++++++++++++++++++++++++++++++++++ channeldb/db_test.go | 85 +++++++++++++++++ 3 files changed, 303 insertions(+) diff --git a/channeldb/channel_test.go b/channeldb/channel_test.go index b1ca100eb..cd7d8db27 100644 --- a/channeldb/channel_test.go +++ b/channeldb/channel_test.go @@ -107,6 +107,10 @@ type testChannelParams struct { // openChannel is set to true if the channel should be fully marked as // open if this is false, the channel will be left in pending state. openChannel bool + + // closedChannel is set to true if the channel should be marked as + // closed after opening it. + closedChannel bool } // testChannelOption is a functional option which can be used to alter the @@ -129,6 +133,21 @@ func openChannelOption() testChannelOption { } } +// closedChannelOption is an option which can be used to create a test channel +// that is closed. +func closedChannelOption() testChannelOption { + return func(params *testChannelParams) { + params.closedChannel = true + } +} + +// pubKeyOption is an option which can be used to set the remote's pubkey. +func pubKeyOption(pubKey *btcec.PublicKey) testChannelOption { + return func(params *testChannelParams) { + params.channel.IdentityPub = pubKey + } +} + // localHtlcsOption is an option which allows setting of htlcs on the local // commitment. func localHtlcsOption(htlcs []HTLC) testChannelOption { @@ -231,6 +250,17 @@ func createTestChannel(t *testing.T, cdb *ChannelStateDB, err = params.channel.MarkAsOpen(params.channel.ShortChannelID) require.NoError(t, err, "unable to mark channel open") + if params.closedChannel { + // Set the other public keys so that serialization doesn't + // panic. + err = params.channel.CloseChannel(&ChannelCloseSummary{ + RemotePub: params.channel.IdentityPub, + RemoteCurrentRevocation: params.channel.IdentityPub, + RemoteNextRevocation: params.channel.IdentityPub, + }) + require.NoError(t, err, "unable to close channel") + } + return params.channel } diff --git a/channeldb/db.go b/channeldb/db.go index bf7909ba5..6b92f2287 100644 --- a/channeldb/db.go +++ b/channeldb/db.go @@ -730,6 +730,194 @@ func (c *ChannelStateDB) FetchChannelByID(tx kvdb.RTx, id lnwire.ChannelID) ( return c.channelScanner(tx, selector) } +// ChanCount is used by the server in determining access control. +type ChanCount struct { + HasOpenOrClosedChan bool + PendingOpenCount uint64 +} + +// FetchPermAndTempPeers returns a map where the key is the remote node's +// public key and the value is a struct that has a tally of the pending-open +// channels and whether the peer has an open or closed channel with us. +func (c *ChannelStateDB) FetchPermAndTempPeers( + chainHash []byte) (map[string]ChanCount, error) { + + peerCounts := make(map[string]ChanCount) + + err := kvdb.View(c.backend, func(tx kvdb.RTx) error { + openChanBucket := tx.ReadBucket(openChannelBucket) + if openChanBucket == nil { + return ErrNoChanDBExists + } + + openChanErr := openChanBucket.ForEach(func(nodePub, + v []byte) error { + + // If there is a value, this is not a bucket. + if v != nil { + return nil + } + + nodeChanBucket := openChanBucket.NestedReadBucket( + nodePub, + ) + if nodeChanBucket == nil { + return nil + } + + chainBucket := nodeChanBucket.NestedReadBucket( + chainHash, + ) + if chainBucket == nil { + return fmt.Errorf("no chain bucket exists") + } + + var isPermPeer bool + var pendingOpenCount uint64 + + internalErr := chainBucket.ForEach(func(chanPoint, + val []byte) error { + + // If there is a value, this is not a bucket. + if val != nil { + return nil + } + + chanBucket := chainBucket.NestedReadBucket( + chanPoint, + ) + if chanBucket == nil { + return nil + } + + var op wire.OutPoint + readErr := graphdb.ReadOutpoint( + bytes.NewReader(chanPoint), &op, + ) + if readErr != nil { + return readErr + } + + // We need to go through each channel and look + // at the IsPending status. + openChan, err := fetchOpenChannel( + chanBucket, &op, + ) + if err != nil { + return err + } + + if openChan.IsPending { + // Add to the pending-open count since + // this is a temp peer. + pendingOpenCount++ + return nil + } + + // Since IsPending is false, this is a perm + // peer. + isPermPeer = true + + return nil + }) + if internalErr != nil { + return internalErr + } + + peerCount := ChanCount{ + HasOpenOrClosedChan: isPermPeer, + PendingOpenCount: pendingOpenCount, + } + peerCounts[string(nodePub)] = peerCount + + return nil + }) + if openChanErr != nil { + return openChanErr + } + + // Now check the closed channel bucket. + historicalChanBucket := tx.ReadBucket(historicalChannelBucket) + if historicalChanBucket == nil { + return ErrNoHistoricalBucket + } + + historicalErr := historicalChanBucket.ForEach(func(chanPoint, + v []byte) error { + // Parse each nested bucket and the chanInfoKey to get + // the IsPending bool. This determines whether the + // peer is protected or not. + if v != nil { + // This is not a bucket. This is currently not + // possible. + return nil + } + + chanBucket := historicalChanBucket.NestedReadBucket( + chanPoint, + ) + if chanBucket == nil { + // This is not possible. + return fmt.Errorf("no historical channel " + + "bucket exists") + } + + var op wire.OutPoint + readErr := graphdb.ReadOutpoint( + bytes.NewReader(chanPoint), &op, + ) + if readErr != nil { + return readErr + } + + // This channel is closed, but the structure of the + // historical bucket is the same. This is by design, + // which means we can call fetchOpenChannel. + channel, fetchErr := fetchOpenChannel(chanBucket, &op) + if fetchErr != nil { + return fetchErr + } + + // Only include this peer in the protected class if + // the closing transaction confirmed. Note that + // CloseChannel can be called in the funding manager + // while IsPending is true which is why we need this + // special-casing to not count premature funding + // manager calls to CloseChannel. + if !channel.IsPending { + // Fetch the public key of the remote node. We + // need to use the string-ified serialized, + // compressed bytes as the key. + remotePub := channel.IdentityPub + remoteSer := remotePub.SerializeCompressed() + remoteKey := string(remoteSer) + + count, exists := peerCounts[remoteKey] + if exists { + count.HasOpenOrClosedChan = true + peerCounts[remoteKey] = count + } else { + peerCount := ChanCount{ + HasOpenOrClosedChan: true, + } + peerCounts[remoteKey] = peerCount + } + } + + return nil + }) + if historicalErr != nil { + return historicalErr + } + + return nil + }, func() { + clear(peerCounts) + }) + + return peerCounts, err +} + // channelSelector describes a function that takes a chain-hash bucket from // within the open-channel DB and returns the wanted channel point bytes, and // channel point. It must return the ErrChannelNotFound error if the wanted diff --git a/channeldb/db_test.go b/channeldb/db_test.go index 9fed9934b..e175ea1fb 100644 --- a/channeldb/db_test.go +++ b/channeldb/db_test.go @@ -721,6 +721,91 @@ func TestFetchHistoricalChannel(t *testing.T) { } } +// TestFetchPermTempPeer tests that we're able to call FetchPermAndTempPeers +// successfully. +func TestFetchPermTempPeer(t *testing.T) { + t.Parallel() + + fullDB, err := MakeTestDB(t) + require.NoError(t, err, "unable to make test database") + + cdb := fullDB.ChannelStateDB() + + // Create an open channel. + privKey1, err := btcec.NewPrivateKey() + require.NoError(t, err, "unable to generate new private key") + + pubKey1 := privKey1.PubKey() + + channelState1 := createTestChannel( + t, cdb, openChannelOption(), pubKeyOption(pubKey1), + ) + + // Next, assert that the channel exists in the database. + _, err = cdb.FetchChannel(channelState1.FundingOutpoint) + require.NoError(t, err, "unable to fetch channel") + + // Create a pending channel. + privKey2, err := btcec.NewPrivateKey() + require.NoError(t, err, "unable to generate private key") + + pubKey2 := privKey2.PubKey() + channelState2 := createTestChannel(t, cdb, pubKeyOption(pubKey2)) + + // Assert that the channel exists in the database. + _, err = cdb.FetchChannel(channelState2.FundingOutpoint) + require.NoError(t, err, "unable to fetch channel") + + // Create a closed channel. + privKey3, err := btcec.NewPrivateKey() + require.NoError(t, err, "unable to generate new private key") + + pubKey3 := privKey3.PubKey() + + _ = createTestChannel( + t, cdb, pubKeyOption(pubKey3), openChannelOption(), + closedChannelOption(), + ) + + // Fetch the ChanCount for our peers. + peerCounts, err := cdb.FetchPermAndTempPeers(key[:]) + require.NoError(t, err, "unable to fetch perm and temp peers") + + // There should only be three entries. + require.Len(t, peerCounts, 3) + + // The first entry should have OpenClosed set to true and Pending set + // to 0. + count1, found := peerCounts[string(pubKey1.SerializeCompressed())] + require.True(t, found, "unable to find peer 1 in peerCounts") + require.True( + t, count1.HasOpenOrClosedChan, + "couldn't find peer 1's channels", + ) + require.Zero( + t, count1.PendingOpenCount, + "peer 1 doesn't have 0 pending-open", + ) + + count2, found := peerCounts[string(pubKey2.SerializeCompressed())] + require.True(t, found, "unable to find peer 2 in peerCounts") + require.False( + t, count2.HasOpenOrClosedChan, "found erroneous channels", + ) + require.Equal(t, uint64(1), count2.PendingOpenCount) + + count3, found := peerCounts[string(pubKey3.SerializeCompressed())] + require.True(t, found, "unable to find peer 3 in peerCounts") + require.True( + t, count3.HasOpenOrClosedChan, + "couldn't find peer 3's channels", + ) + require.Zero( + t, count3.PendingOpenCount, + "peer 3 doesn't have 0 pending-open", + ) +} + func createLightningNode(priv *btcec.PrivateKey) *models.LightningNode { updateTime := rand.Int63() From 4cfc92f420b7dea5a9cb95ba144b188ff73bb6da Mon Sep 17 00:00:00 2001 From: Eugene Siegel Date: Wed, 15 Jan 2025 13:17:48 -0500 Subject: [PATCH 2/7] channelnotifier: add FundingTimeout and NotifyFundingTimeout This signal will be used in the server.go code to potentially demote temporary-access peers to restricted-access peers. --- channelnotifier/channelnotifier.go | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/channelnotifier/channelnotifier.go b/channelnotifier/channelnotifier.go index b03d91313..0019e50f9 100644 --- a/channelnotifier/channelnotifier.go +++ b/channelnotifier/channelnotifier.go @@ -80,6 +80,14 @@ type FullyResolvedChannelEvent struct { ChannelPoint *wire.OutPoint } +// FundingTimeoutEvent represents a new event where a pending-open channel has +// timed out from the PoV of the funding manager because the funding tx +// has not confirmed in the allotted time. +type FundingTimeoutEvent struct { + // ChannelPoint is the channelpoint for the newly inactive channel. + ChannelPoint *wire.OutPoint +} + // New creates a new channel notifier. The ChannelNotifier gets channel // events from peers and from the chain arbitrator, and dispatches them to // its clients. @@ -184,6 +192,17 @@ func (c *ChannelNotifier) NotifyFullyResolvedChannelEvent( } } +// NotifyFundingTimeoutEvent notifies the channelEventNotifier goroutine that +// a funding timeout has occurred for a certain channel point. +func (c *ChannelNotifier) NotifyFundingTimeout(chanPoint wire.OutPoint) { + // Send this event to all channel event subscribers. + event := FundingTimeoutEvent{ChannelPoint: &chanPoint} + if err := c.ntfnServer.SendUpdate(event); err != nil { + log.Warnf("Unable to send funding timeout update: %v for "+ + "ChanPoint(%v)", err, chanPoint) + } +} + // NotifyActiveLinkEvent notifies the channelEventNotifier goroutine that a // link has been added to the switch. func (c *ChannelNotifier) NotifyActiveLinkEvent(chanPoint wire.OutPoint) { From 6eb746fbbab384914429bf857f64659abc657453 Mon Sep 17 00:00:00 2001 From: Eugene Siegel Date: Thu, 16 Jan 2025 10:32:34 -0500 Subject: [PATCH 3/7] server.go+accessman.go: introduce caches for access permissions Here we introduce the access manager which has caches that will determine the access control status of our peers. Peers that have had their funding transaction confirm with us are protected. Peers that only have pending-open channels with us are temporary access and can have their access revoked. The rest of the peers are granted restricted access. --- accessman.go | 384 +++++++++++++++++++++++++++++++++++++++ accessman_test.go | 153 ++++++++++++++++ brontide/listener.go | 40 +++- brontide/noise_test.go | 2 +- config.go | 15 +- discovery/syncer.go | 7 +- funding/manager.go | 2 +- peer/brontide.go | 7 +- server.go | 230 ++++++++++++++++++----- watchtower/standalone.go | 1 + 10 files changed, 788 insertions(+), 53 deletions(-) create mode 100644 accessman.go create mode 100644 accessman_test.go diff --git a/accessman.go b/accessman.go new file mode 100644 index 000000000..e1ccc5c34 --- /dev/null +++ b/accessman.go @@ -0,0 +1,384 @@ +package lnd + +import ( + "fmt" + "sync" + + "github.com/btcsuite/btcd/btcec/v2" + "github.com/lightningnetwork/lnd/channeldb" +) + +// accessMan is responsible for managing the server's access permissions. +type accessMan struct { + cfg *accessManConfig + + // banScoreMtx is used for the server's ban tracking. If the server + // mutex is also going to be locked, ensure that this is locked after + // the server mutex. + banScoreMtx sync.RWMutex + + // peerCounts is a mapping from remote public key to {bool, uint64} + // where the bool indicates that we have an open/closed channel with + // the peer and where the uint64 indicates the number of pending-open + // channels we currently have with them. This mapping will be used to + // determine access permissions for the peer. The map key is the + // string-version of the serialized public key. + // + // NOTE: This MUST be accessed with the banScoreMtx held. + peerCounts map[string]channeldb.ChanCount + + // peerScores stores each connected peer's access status. The map key + // is the string-version of the serialized public key. + // + // NOTE: This MUST be accessed with the banScoreMtx held. + peerScores map[string]peerSlotStatus + + // numRestricted tracks the number of peers with restricted access in + // peerScores. This MUST be accessed with the banScoreMtx held. + numRestricted int64 +} + +type accessManConfig struct { + // initAccessPerms checks the channeldb for initial access permissions + // and then populates the peerCounts and peerScores maps. + initAccessPerms func() (map[string]channeldb.ChanCount, error) + + // shouldDisconnect determines whether we should disconnect a peer or + // not. + shouldDisconnect func(*btcec.PublicKey) (bool, error) + + // maxRestrictedSlots is the number of restricted slots we'll allocate. + maxRestrictedSlots int64 +} + +func newAccessMan(cfg *accessManConfig) (*accessMan, error) { + a := &accessMan{ + cfg: cfg, + peerCounts: make(map[string]channeldb.ChanCount), + peerScores: make(map[string]peerSlotStatus), + } + + counts, err := a.cfg.initAccessPerms() + if err != nil { + return nil, err + } + + // We'll populate the server's peerCounts map with the counts fetched + // via initAccessPerms. Also note that we haven't yet connected to the + // peers. + for peerPub, count := range counts { + a.peerCounts[peerPub] = count + } + + return a, nil +} + +// assignPeerPerms assigns a new peer its permissions. This does not track the +// access in the maps. This is intentional. +func (a *accessMan) assignPeerPerms(remotePub *btcec.PublicKey) ( + peerAccessStatus, error) { + + // Default is restricted unless the below filters say otherwise. + access := peerStatusRestricted + + shouldDisconnect, err := a.cfg.shouldDisconnect(remotePub) + if err != nil { + // Access is restricted here. + return access, err + } + + if shouldDisconnect { + // Access is restricted here. + return access, ErrGossiperBan + } + + peerMapKey := string(remotePub.SerializeCompressed()) + + // Lock banScoreMtx for reading so that we can update the banning maps + // below. + a.banScoreMtx.RLock() + defer a.banScoreMtx.RUnlock() + + if count, found := a.peerCounts[peerMapKey]; found { + if count.HasOpenOrClosedChan { + access = peerStatusProtected + } else if count.PendingOpenCount != 0 { + access = peerStatusTemporary + } + } + + // If we've reached this point and access hasn't changed from + // restricted, then we need to check if we even have a slot for this + // peer. + if a.numRestricted >= a.cfg.maxRestrictedSlots && + access == peerStatusRestricted { + + return access, ErrNoMoreRestrictedAccessSlots + } + + return access, nil +} + +// newPendingOpenChan is called after the pending-open channel has been +// committed to the database. This may transition a restricted-access peer to a +// temporary-access peer. +func (a *accessMan) newPendingOpenChan(remotePub *btcec.PublicKey) error { + a.banScoreMtx.Lock() + defer a.banScoreMtx.Unlock() + + peerMapKey := string(remotePub.SerializeCompressed()) + + // Fetch the peer's access status from peerScores. + status, found := a.peerScores[peerMapKey] + if !found { + // If we didn't find the peer, we'll return an error. + return ErrNoPeerScore + } + + switch status.state { + case peerStatusProtected: + // If this peer's access status is protected, we don't need to + // do anything. + return nil + + case peerStatusTemporary: + // If this peer's access status is temporary, we'll need to + // update the peerCounts map. The peer's access status will + // stay temporary. + peerCount, found := a.peerCounts[peerMapKey] + if !found { + // Error if we did not find any info in peerCounts. + return ErrNoPendingPeerInfo + } + + // Increment the pending channel amount. + peerCount.PendingOpenCount += 1 + a.peerCounts[peerMapKey] = peerCount + + case peerStatusRestricted: + // If the peer's access status is restricted, then we can + // transition it to a temporary-access peer. We'll need to + // update numRestricted and also peerScores. We'll also need to + // update peerCounts. + peerCount := channeldb.ChanCount{ + HasOpenOrClosedChan: false, + PendingOpenCount: 1, + } + + a.peerCounts[peerMapKey] = peerCount + + // A restricted-access slot has opened up. + a.numRestricted -= 1 + + a.peerScores[peerMapKey] = peerSlotStatus{ + state: peerStatusTemporary, + } + + default: + // This should not be possible. + return fmt.Errorf("invalid peer access status") + } + + return nil +} + +// newPendingCloseChan is called when a pending-open channel prematurely closes +// before the funding transaction has confirmed. This potentially demotes a +// temporary-access peer to a restricted-access peer. If no restricted-access +// slots are available, the peer will be disconnected. +func (a *accessMan) newPendingCloseChan(remotePub *btcec.PublicKey) error { + a.banScoreMtx.Lock() + defer a.banScoreMtx.Unlock() + + peerMapKey := string(remotePub.SerializeCompressed()) + + // Fetch the peer's access status from peerScores. + status, found := a.peerScores[peerMapKey] + if !found { + return ErrNoPeerScore + } + + switch status.state { + case peerStatusProtected: + // If this peer is protected, we don't do anything. + return nil + + case peerStatusTemporary: + // If this peer is temporary, we need to check if it will + // revert to a restricted-access peer. + peerCount, found := a.peerCounts[peerMapKey] + if !found { + // Error if we did not find any info in peerCounts. + return ErrNoPendingPeerInfo + } + + currentNumPending := peerCount.PendingOpenCount - 1 + if currentNumPending == 0 { + // Remove the entry from peerCounts. + delete(a.peerCounts, peerMapKey) + + // If this is the only pending-open channel for this + // peer and it's getting removed, attempt to demote + // this peer to a restricted peer. + if a.numRestricted == a.cfg.maxRestrictedSlots { + // There are no available restricted slots, so + // we need to disconnect this peer. We leave + // this up to the caller. + return ErrNoMoreRestrictedAccessSlots + } + + // Otherwise, there is an available restricted-access + // slot, so we can demote this peer. + a.peerScores[peerMapKey] = peerSlotStatus{ + state: peerStatusRestricted, + } + + // Update numRestricted. + a.numRestricted++ + + return nil + } + + // Else, we don't need to demote this peer since it has other + // pending-open channels with us. + peerCount.PendingOpenCount = currentNumPending + a.peerCounts[peerMapKey] = peerCount + + return nil + + case peerStatusRestricted: + // This should not be possible. This indicates an error. + return fmt.Errorf("invalid peer access state transition") + + default: + // This should not be possible. + return fmt.Errorf("invalid peer access status") + } +} + +// newOpenChan is called when a pending-open channel becomes an open channel +// (i.e. the funding transaction has confirmed). If the remote peer is a +// temporary-access peer, it will be promoted to a protected-access peer. +func (a *accessMan) newOpenChan(remotePub *btcec.PublicKey) error { + a.banScoreMtx.Lock() + defer a.banScoreMtx.Unlock() + + peerMapKey := string(remotePub.SerializeCompressed()) + + // Fetch the peer's access status from peerScores. + status, found := a.peerScores[peerMapKey] + if !found { + // If we didn't find the peer, we'll return an error. + return ErrNoPeerScore + } + + switch status.state { + case peerStatusProtected: + // If the peer's state is already protected, we don't need to + // do anything more. + return nil + + case peerStatusTemporary: + // If the peer's state is temporary, we'll upgrade the peer to + // a protected peer. + peerCount, found := a.peerCounts[peerMapKey] + if !found { + // Error if we did not find any info in peerCounts. + return ErrNoPendingPeerInfo + } + + peerCount.HasOpenOrClosedChan = true + a.peerCounts[peerMapKey] = peerCount + + newStatus := peerSlotStatus{ + state: peerStatusProtected, + } + a.peerScores[peerMapKey] = newStatus + + return nil + + case peerStatusRestricted: + // This should not be possible. For the server to receive a + // state-transition event via NewOpenChan, the server must have + // previously granted this peer "temporary" access. This + // temporary access would not have been revoked or downgraded + // without `CloseChannel` being called with the pending + // argument set to true. This means that an open-channel state + // transition would be impossible. Therefore, we can return an + // error. + return fmt.Errorf("invalid peer access status") + + default: + // This should not be possible. + return fmt.Errorf("invalid peer access status") + } +} + +// checkIncomingConnBanScore checks whether, given the remote's public hex- +// encoded key, we should not accept this incoming connection or immediately +// disconnect. This does not assign to the server's peerScores maps. This is +// just an inbound filter that the brontide listeners use. +func (a *accessMan) checkIncomingConnBanScore(remotePub *btcec.PublicKey) ( + bool, error) { + + a.banScoreMtx.RLock() + defer a.banScoreMtx.RUnlock() + + peerMapKey := string(remotePub.SerializeCompressed()) + + if _, found := a.peerCounts[peerMapKey]; !found { + // Check numRestricted to see if there is an available slot. In + // the future, it's possible to add better heuristics. + if a.numRestricted < a.cfg.maxRestrictedSlots { + // There is an available slot. + return true, nil + } + + // If there are no slots left, then we reject this connection. + return false, ErrNoMoreRestrictedAccessSlots + } + + // Else, the peer is either protected or temporary. + return true, nil +} + +// addPeerAccess tracks a peer's access in the maps. This should be called when +// the peer has fully connected. +func (a *accessMan) addPeerAccess(remotePub *btcec.PublicKey, + access peerAccessStatus) { + + // Add the remote public key to peerScores. + a.banScoreMtx.Lock() + defer a.banScoreMtx.Unlock() + + peerMapKey := string(remotePub.SerializeCompressed()) + + a.peerScores[peerMapKey] = peerSlotStatus{state: access} + + // Increment numRestricted. + if access == peerStatusRestricted { + a.numRestricted++ + } +} + +// removePeerAccess removes the peer's access from the maps. This should be +// called when the peer has been disconnected. +func (a *accessMan) removePeerAccess(remotePub *btcec.PublicKey) { + a.banScoreMtx.Lock() + defer a.banScoreMtx.Unlock() + + peerMapKey := string(remotePub.SerializeCompressed()) + + status, found := a.peerScores[peerMapKey] + if !found { + return + } + + if status.state == peerStatusRestricted { + // If the status is restricted, then we decrement from + // numRestrictedSlots. + a.numRestricted-- + } + + delete(a.peerScores, peerMapKey) +} diff --git a/accessman_test.go b/accessman_test.go new file mode 100644 index 000000000..ee638cf13 --- /dev/null +++ b/accessman_test.go @@ -0,0 +1,153 @@ +package lnd + +import ( + "testing" + + "github.com/btcsuite/btcd/btcec/v2" + "github.com/lightningnetwork/lnd/channeldb" + "github.com/stretchr/testify/require" +) + +// assertInboundConnection asserts that we're able to accept an inbound +// connection successfully without any access permissions being violated. +func assertInboundConnection(t *testing.T, a *accessMan, + remotePub *btcec.PublicKey, status peerAccessStatus) { + + remotePubSer := string(remotePub.SerializeCompressed()) + + isSlotAvailable, err := a.checkIncomingConnBanScore(remotePub) + require.NoError(t, err) + require.True(t, isSlotAvailable) + + peerAccess, err := a.assignPeerPerms(remotePub) + require.NoError(t, err) + require.Equal(t, status, peerAccess) + + a.addPeerAccess(remotePub, peerAccess) + peerScore, ok := a.peerScores[remotePubSer] + require.True(t, ok) + require.Equal(t, status, peerScore.state) +} + +func assertAccessState(t *testing.T, a *accessMan, remotePub *btcec.PublicKey, + expectedStatus peerAccessStatus) { + + remotePubSer := string(remotePub.SerializeCompressed()) + peerScore, ok := a.peerScores[remotePubSer] + require.True(t, ok) + require.Equal(t, expectedStatus, peerScore.state) +} + +// TestAccessManRestrictedSlots tests that the configurable number of +// restricted slots are properly allocated. It also tests that certain peers +// with access permissions are allowed to bypass the slot mechanism. +func TestAccessManRestrictedSlots(t *testing.T) { + t.Parallel() + + // We'll pre-populate the map to mock the database fetch. We'll make + // three peers. One has an open/closed channel. One has both an open + // / closed channel and a pending channel. The last one has only a + // pending channel. + peerPriv1, err := btcec.NewPrivateKey() + require.NoError(t, err) + peerKey1 := peerPriv1.PubKey() + peerKeySer1 := string(peerKey1.SerializeCompressed()) + + peerPriv2, err := btcec.NewPrivateKey() + require.NoError(t, err) + peerKey2 := peerPriv2.PubKey() + peerKeySer2 := string(peerKey2.SerializeCompressed()) + + peerPriv3, err := btcec.NewPrivateKey() + require.NoError(t, err) + peerKey3 := peerPriv3.PubKey() + peerKeySer3 := string(peerKey3.SerializeCompressed()) + + initPerms := func() (map[string]channeldb.ChanCount, error) { + return map[string]channeldb.ChanCount{ + peerKeySer1: { + HasOpenOrClosedChan: true, + }, + peerKeySer2: { + HasOpenOrClosedChan: true, + PendingOpenCount: 1, + }, + peerKeySer3: { + HasOpenOrClosedChan: false, + PendingOpenCount: 1, + }, + }, nil + } + + disconnect := func(*btcec.PublicKey) (bool, error) { + return false, nil + } + + cfg := &accessManConfig{ + initAccessPerms: initPerms, + shouldDisconnect: disconnect, + maxRestrictedSlots: 1, + } + + a, err := newAccessMan(cfg) + require.NoError(t, err) + + // Check that the peerCounts map is correctly populated with three + // peers. + require.Equal(t, 0, int(a.numRestricted)) + require.Equal(t, 3, len(a.peerCounts)) + + peerCount1, ok := a.peerCounts[peerKeySer1] + require.True(t, ok) + require.True(t, peerCount1.HasOpenOrClosedChan) + require.Equal(t, 0, int(peerCount1.PendingOpenCount)) + + peerCount2, ok := a.peerCounts[peerKeySer2] + require.True(t, ok) + require.True(t, peerCount2.HasOpenOrClosedChan) + require.Equal(t, 1, int(peerCount2.PendingOpenCount)) + + peerCount3, ok := a.peerCounts[peerKeySer3] + require.True(t, ok) + require.False(t, peerCount3.HasOpenOrClosedChan) + require.Equal(t, 1, int(peerCount3.PendingOpenCount)) + + // We'll now start to connect the peers. We'll add a new fourth peer + // that will take up the restricted slot. The first three peers should + // be able to bypass this restricted slot mechanism. + peerPriv4, err := btcec.NewPrivateKey() + require.NoError(t, err) + peerKey4 := peerPriv4.PubKey() + + // Follow the normal process of an incoming connection. We check if we + // can accommodate this peer in checkIncomingConnBanScore and then we + // assign its access permissions and then insert into the map. + assertInboundConnection(t, a, peerKey4, peerStatusRestricted) + + // Connect the three peers. This should happen without any issue. + assertInboundConnection(t, a, peerKey1, peerStatusProtected) + assertInboundConnection(t, a, peerKey2, peerStatusProtected) + assertInboundConnection(t, a, peerKey3, peerStatusTemporary) + + // Check that a pending-open channel promotes the restricted peer. + err = a.newPendingOpenChan(peerKey4) + require.NoError(t, err) + assertAccessState(t, a, peerKey4, peerStatusTemporary) + + // Check that an open channel promotes the temporary peer. + err = a.newOpenChan(peerKey3) + require.NoError(t, err) + assertAccessState(t, a, peerKey3, peerStatusProtected) + + // We should be able to accommodate a new peer. + peerPriv5, err := btcec.NewPrivateKey() + require.NoError(t, err) + peerKey5 := peerPriv5.PubKey() + + assertInboundConnection(t, a, peerKey5, peerStatusRestricted) + + // Check that a pending-close channel event for peer 4 demotes the + // peer. + err = a.newPendingCloseChan(peerKey4) + require.ErrorIs(t, err, ErrNoMoreRestrictedAccessSlots) +} diff --git a/brontide/listener.go b/brontide/listener.go index ba210853c..4d386c1c5 100644 --- a/brontide/listener.go +++ b/brontide/listener.go @@ -7,12 +7,13 @@ import ( "net" "time" + "github.com/btcsuite/btcd/btcec/v2" "github.com/lightningnetwork/lnd/keychain" ) // defaultHandshakes is the maximum number of handshakes that can be done in // parallel. -const defaultHandshakes = 1000 +const defaultHandshakes = 50 // Listener is an implementation of a net.Conn which executes an authenticated // key exchange and message encryption protocol dubbed "Machine" after @@ -24,6 +25,10 @@ type Listener struct { tcp *net.TCPListener + // shouldAccept is a closure that determines if we should accept the + // incoming connection or not based on its public key. + shouldAccept func(*btcec.PublicKey) (bool, error) + handshakeSema chan struct{} conns chan maybeConn quit chan struct{} @@ -34,8 +39,8 @@ var _ net.Listener = (*Listener)(nil) // NewListener returns a new net.Listener which enforces the Brontide scheme // during both initial connection establishment and data transfer. -func NewListener(localStatic keychain.SingleKeyECDH, - listenAddr string) (*Listener, error) { +func NewListener(localStatic keychain.SingleKeyECDH, listenAddr string, + shouldAccept func(*btcec.PublicKey) (bool, error)) (*Listener, error) { addr, err := net.ResolveTCPAddr("tcp", listenAddr) if err != nil { @@ -50,6 +55,7 @@ func NewListener(localStatic keychain.SingleKeyECDH, brontideListener := &Listener{ localStatic: localStatic, tcp: l, + shouldAccept: shouldAccept, handshakeSema: make(chan struct{}, defaultHandshakes), conns: make(chan maybeConn), quit: make(chan struct{}), @@ -193,6 +199,28 @@ func (l *Listener) doHandshake(conn net.Conn) { return } + // Call the shouldAccept closure to see if the remote node's public key + // is allowed according to our banning heuristic. This is here because + // we do not learn the remote node's public static key until we've + // received and validated Act 3. + remoteKey := brontideConn.RemotePub() + if remoteKey == nil { + connErr := fmt.Errorf("no remote pubkey") + brontideConn.conn.Close() + l.rejectConn(rejectedConnErr(connErr, remoteAddr)) + + return + } + + accepted, acceptErr := l.shouldAccept(remoteKey) + if !accepted { + // Reject the connection. + brontideConn.conn.Close() + l.rejectConn(rejectedConnErr(acceptErr, remoteAddr)) + + return + } + l.acceptConn(brontideConn) } @@ -255,3 +283,9 @@ func (l *Listener) Close() error { func (l *Listener) Addr() net.Addr { return l.tcp.Addr() } + +// DisabledBanClosure is used in places that NewListener is invoked to bypass +// the ban-scoring. +func DisabledBanClosure(p *btcec.PublicKey) (bool, error) { + return true, nil +} diff --git a/brontide/noise_test.go b/brontide/noise_test.go index fc4ff99e1..cd4dc4cd9 100644 --- a/brontide/noise_test.go +++ b/brontide/noise_test.go @@ -35,7 +35,7 @@ func makeListener() (*Listener, *lnwire.NetAddress, error) { addr := "localhost:0" // Our listener will be local, and the connection remote. - listener, err := NewListener(localKeyECDH, addr) + listener, err := NewListener(localKeyECDH, addr, DisabledBanClosure) if err != nil { return nil, nil, err } diff --git a/config.go b/config.go index 6418d9a68..1a06ed765 100644 --- a/config.go +++ b/config.go @@ -238,6 +238,10 @@ const ( // defaultHTTPHeaderTimeout is the default timeout for HTTP requests. DefaultHTTPHeaderTimeout = 5 * time.Second + // DefaultNumRestrictedSlots is the default number of restricted slots + // we'll allocate in the server. + DefaultNumRestrictedSlots = 30 + // BitcoinChainName is a string that represents the Bitcoin blockchain. BitcoinChainName = "bitcoin" @@ -518,6 +522,10 @@ type Config struct { // HTTPHeaderTimeout is the maximum duration that the server will wait // before timing out reading the headers of an HTTP request. HTTPHeaderTimeout time.Duration `long:"http-header-timeout" description:"The maximum duration that the server will wait before timing out reading the headers of an HTTP request."` + + // NumRestrictedSlots is the number of restricted slots we'll allocate + // in the server. + NumRestrictedSlots uint64 `long:"num-restricted-slots" description:"The number of restricted slots we'll allocate in the server."` } // GRPCConfig holds the configuration options for the gRPC server. @@ -735,9 +743,10 @@ func DefaultConfig() Config { ServerPingTimeout: defaultGrpcServerPingTimeout, ClientPingMinWait: defaultGrpcClientPingMinWait, }, - LogConfig: build.DefaultLogConfig(), - WtClient: lncfg.DefaultWtClientCfg(), - HTTPHeaderTimeout: DefaultHTTPHeaderTimeout, + LogConfig: build.DefaultLogConfig(), + WtClient: lncfg.DefaultWtClientCfg(), + HTTPHeaderTimeout: DefaultHTTPHeaderTimeout, + NumRestrictedSlots: DefaultNumRestrictedSlots, } } diff --git a/discovery/syncer.go b/discovery/syncer.go index 6b19908d5..991bd75d3 100644 --- a/discovery/syncer.go +++ b/discovery/syncer.go @@ -181,6 +181,9 @@ const ( // requestBatchSize is the maximum number of channels we will query the // remote peer for in a QueryShortChanIDs message. requestBatchSize = 500 + + // syncerBufferSize is the size of the syncer's buffers. + syncerBufferSize = 5 ) var ( @@ -436,8 +439,8 @@ func newGossipSyncer(cfg gossipSyncerCfg, sema chan struct{}) *GossipSyncer { rateLimiter: rateLimiter, syncTransitionReqs: make(chan *syncTransitionReq), historicalSyncReqs: make(chan *historicalSyncReq), - gossipMsgs: make(chan lnwire.Message, 100), - queryMsgs: make(chan lnwire.Message, 100), + gossipMsgs: make(chan lnwire.Message, syncerBufferSize), + queryMsgs: make(chan lnwire.Message, syncerBufferSize), syncerSema: sema, quit: make(chan struct{}), } diff --git a/funding/manager.go b/funding/manager.go index 6e81c6f8b..86f49e0eb 100644 --- a/funding/manager.go +++ b/funding/manager.go @@ -105,7 +105,7 @@ const ( // pendingChansLimit is the maximum number of pending channels that we // can have. After this point, pending channel opens will start to be // rejected. - pendingChansLimit = 1_000 + pendingChansLimit = 50 ) var ( diff --git a/peer/brontide.go b/peer/brontide.go index f0399b4e8..2e09aeaad 100644 --- a/peer/brontide.go +++ b/peer/brontide.go @@ -90,6 +90,9 @@ const ( // torTimeoutMultiplier is the scaling factor we use on network timeouts // for Tor peers. torTimeoutMultiplier = 3 + + // msgStreamSize is the size of the message streams. + msgStreamSize = 5 ) var ( @@ -1856,7 +1859,7 @@ func newChanMsgStream(p *Brontide, cid lnwire.ChannelID) *msgStream { return newMsgStream(p, fmt.Sprintf("Update stream for ChannelID(%x) created", cid[:]), fmt.Sprintf("Update stream for ChannelID(%x) exiting", cid[:]), - 1000, + msgStreamSize, apply, ) } @@ -1875,7 +1878,7 @@ func newDiscMsgStream(p *Brontide) *msgStream { p, "Update stream for gossiper created", "Update stream for gossiper exited", - 1000, + msgStreamSize, apply, ) } diff --git a/server.go b/server.go index 01019583f..65672c9e1 100644 --- a/server.go +++ b/server.go @@ -140,6 +140,25 @@ var ( // // Per blip04: January 1, 2026 12:00:00 AM UTC in unix seconds. EndorsementExperimentEnd = time.Unix(1767225600, 0) + + // ErrGossiperBan is one of the errors that can be returned when we + // attempt to finalize a connection to a remote peer. + ErrGossiperBan = errors.New("gossiper has banned remote's key") + + // ErrNoMoreRestrictedAccessSlots is one of the errors that can be + // returned when we attempt to finalize a connection. It means that + // this peer has no pending-open, open, or closed channels with us and + // are already at our connection ceiling for a peer with this access + // status. + ErrNoMoreRestrictedAccessSlots = errors.New("no more restricted slots") + + // ErrNoPeerScore is returned when we expect to find a score in + // peerScores, but one does not exist. + ErrNoPeerScore = errors.New("peer score not found") + + // ErrNoPendingPeerInfo is returned when we couldn't find any pending + // peer info. + ErrNoPendingPeerInfo = errors.New("no pending peer info") ) // errPeerAlreadyConnected is an error returned by the server when we're @@ -155,6 +174,33 @@ func (e *errPeerAlreadyConnected) Error() string { return fmt.Sprintf("already connected to peer: %v", e.peer) } +// peerAccessStatus denotes the p2p access status of a given peer. This will be +// used to assign peer ban scores that determine an action the server will +// take. +type peerAccessStatus int + +const ( + // peerStatusRestricted indicates that the peer only has access to the + // limited number of "free" reserved slots. + peerStatusRestricted peerAccessStatus = iota + + // peerStatusTemporary indicates that the peer only has temporary p2p + // access to the server. + peerStatusTemporary + + // peerStatusProtected indicates that the peer has been granted + // permanent p2p access to the server. The peer can still have its + // access revoked. + peerStatusProtected +) + +// peerSlotStatus determines whether a peer gets access to one of our free +// slots or gets to bypass this safety mechanism. +type peerSlotStatus struct { + // state determines which privileges the peer has with our server. + state peerAccessStatus +} + // server is the main server of the Lightning Network Daemon. The server houses // global state pertaining to the wallet, database, and the rpcserver. // Additionally, the server is also used as a central messaging bus to interact @@ -360,6 +406,9 @@ type server struct { // of new blocks. blockbeatDispatcher *chainio.BlockbeatDispatcher + // peerAccessMan implements peer access controls. + peerAccessMan *accessMan + quit chan struct{} wg sync.WaitGroup @@ -525,19 +574,6 @@ func newServer(cfg *Config, listenAddrs []net.Addr, ) ) - listeners := make([]net.Listener, len(listenAddrs)) - for i, listenAddr := range listenAddrs { - // Note: though brontide.NewListener uses ResolveTCPAddr, it - // doesn't need to call the general lndResolveTCP function - // since we are resolving a local address. - listeners[i], err = brontide.NewListener( - nodeKeyECDH, listenAddr.String(), - ) - if err != nil { - return nil, err - } - } - var serializedPubKey [33]byte copy(serializedPubKey[:], nodeKeyDesc.PubKey.SerializeCompressed()) @@ -1142,6 +1178,26 @@ func newServer(cfg *Config, listenAddrs []net.Addr, AssumeChannelValid: cfg.Routing.AssumeChannelValid, }, nodeKeyDesc) + accessCfg := &accessManConfig{ + initAccessPerms: func() (map[string]channeldb.ChanCount, + error) { + + genesisHash := *s.cfg.ActiveNetParams.GenesisHash + return s.chanStateDB.FetchPermAndTempPeers( + genesisHash[:], + ) + }, + shouldDisconnect: s.authGossiper.ShouldDisconnect, + maxRestrictedSlots: int64(s.cfg.NumRestrictedSlots), + } + + peerAccessMan, err := newAccessMan(accessCfg) + if err != nil { + return nil, err + } + + s.peerAccessMan = peerAccessMan + selfVertex := route.Vertex(nodeKeyDesc.PubKey.SerializeCompressed()) //nolint:ll s.localChanMgr = &localchans.Manager{ @@ -1816,6 +1872,23 @@ func newServer(cfg *Config, listenAddrs []net.Addr, // Create liveness monitor. s.createLivenessMonitor(cfg, cc, leaderElector) + listeners := make([]net.Listener, len(listenAddrs)) + for i, listenAddr := range listenAddrs { + // Note: though brontide.NewListener uses ResolveTCPAddr, it + // doesn't need to call the general lndResolveTCP function + // since we are resolving a local address. + + // RESOLVE: We are actually partially accepting inbound + // connection requests when we call NewListener. + listeners[i], err = brontide.NewListener( + nodeKeyECDH, listenAddr.String(), + s.peerAccessMan.checkIncomingConnBanScore, + ) + if err != nil { + return nil, err + } + } + // Create the connection manager which will be responsible for // maintaining persistent outbound connections and also accepting new // incoming connections @@ -2397,6 +2470,11 @@ func (s *server) Start() error { s.connMgr.Stop() return nil }) + + // RESOLVE: s.connMgr.Start() is called here, but + // brontide.NewListener() is called in newServer. This means + // that we are actually listening and partially accepting + // inbound connections even before the connMgr starts. s.connMgr.Start() // If peers are specified as a config option, we'll add those @@ -3622,6 +3700,23 @@ func (s *server) prunePersistentPeerConnection(compressedPubKey [33]byte) { s.mu.Unlock() } +// bannedPersistentPeerConnection does not actually "ban" a persistent peer. It +// is instead used to remove persistent peer state for a peer that has been +// disconnected for good cause by the server. Currently, a gossip ban from +// sending garbage and the server running out of restricted-access +// (i.e. "free") connection slots are the only way this logic gets hit. In the +// future, this function may expand when more ban criteria is added. +// +// NOTE: The server's write lock MUST be held when this is called. +func (s *server) bannedPersistentPeerConnection(remotePub string) { + if perm, ok := s.persistentPeers[remotePub]; ok && !perm { + delete(s.persistentPeers, remotePub) + delete(s.persistentPeersBackoff, remotePub) + delete(s.persistentPeerAddrs, remotePub) + s.cancelConnReqs(remotePub, nil) + } +} + // BroadcastMessage sends a request to the server to broadcast a set of // messages to all peers other than the one specified by the `skips` parameter. // All messages sent via BroadcastMessage will be queued for lazy delivery to @@ -3875,18 +3970,15 @@ func (s *server) InboundPeerConnected(conn net.Conn) { defer s.mu.Unlock() // If the remote node's public key is banned, drop the connection. - shouldDc, dcErr := s.authGossiper.ShouldDisconnect(nodePub) - if dcErr != nil { - srvrLog.Errorf("Unable to check if we should disconnect "+ - "peer: %v", dcErr) - conn.Close() + access, err := s.peerAccessMan.assignPeerPerms(nodePub) + if err != nil { + // Clean up the persistent peer maps if we're dropping this + // connection. + s.bannedPersistentPeerConnection(pubStr) - return - } - - if shouldDc { - srvrLog.Debugf("Dropping connection for %v since they are "+ - "banned.", pubSer) + srvrLog.Debugf("Dropping connection for %x since we are out "+ + "of restricted-access connection slots: %v.", pubSer, + err) conn.Close() @@ -3927,7 +4019,7 @@ func (s *server) InboundPeerConnected(conn net.Conn) { // We were unable to locate an existing connection with the // target peer, proceed to connect. s.cancelConnReqs(pubStr, nil) - s.peerConnected(conn, nil, true) + s.peerConnected(conn, nil, true, access) case nil: // We already have a connection with the incoming peer. If the @@ -3959,7 +4051,7 @@ func (s *server) InboundPeerConnected(conn net.Conn) { s.removePeer(connectedPeer) s.ignorePeerTermination[connectedPeer] = struct{}{} s.scheduledPeerConnection[pubStr] = func() { - s.peerConnected(conn, nil, true) + s.peerConnected(conn, nil, true, access) } } } @@ -3984,19 +4076,15 @@ func (s *server) OutboundPeerConnected(connReq *connmgr.ConnReq, conn net.Conn) s.mu.Lock() defer s.mu.Unlock() - // If the remote node's public key is banned, drop the connection. - shouldDc, dcErr := s.authGossiper.ShouldDisconnect(nodePub) - if dcErr != nil { - srvrLog.Errorf("Unable to check if we should disconnect "+ - "peer: %v", dcErr) - conn.Close() + access, err := s.peerAccessMan.assignPeerPerms(nodePub) + if err != nil { + // Clean up the persistent peer maps if we're dropping this + // connection. + s.bannedPersistentPeerConnection(pubStr) - return - } - - if shouldDc { - srvrLog.Debugf("Dropping connection for %v since they are "+ - "banned.", pubSer) + srvrLog.Debugf("Dropping connection for %x since we are out "+ + "of restricted-access connection slots: %v.", pubSer, + err) if connReq != nil { s.connMgr.Remove(connReq.ID()) @@ -4065,7 +4153,7 @@ func (s *server) OutboundPeerConnected(connReq *connmgr.ConnReq, conn net.Conn) case ErrPeerNotConnected: // We were unable to locate an existing connection with the // target peer, proceed to connect. - s.peerConnected(conn, connReq, false) + s.peerConnected(conn, connReq, false, access) case nil: // We already have a connection with the incoming peer. If the @@ -4099,7 +4187,7 @@ func (s *server) OutboundPeerConnected(connReq *connmgr.ConnReq, conn net.Conn) s.removePeer(connectedPeer) s.ignorePeerTermination[connectedPeer] = struct{}{} s.scheduledPeerConnection[pubStr] = func() { - s.peerConnected(conn, connReq, false) + s.peerConnected(conn, connReq, false, access) } } } @@ -4173,12 +4261,66 @@ func (s *server) SubscribeCustomMessages() (*subscribe.Client, error) { return s.customMessageServer.Subscribe() } +// notifyOpenChannelPeerEvent updates the access manager's maps and then calls +// the channelNotifier's NotifyOpenChannelEvent. +func (s *server) notifyOpenChannelPeerEvent(op wire.OutPoint, + remotePub *btcec.PublicKey) error { + + // Call newOpenChan to update the access manager's maps for this peer. + if err := s.peerAccessMan.newOpenChan(remotePub); err != nil { + return err + } + + // Notify subscribers about this open channel event. + s.channelNotifier.NotifyOpenChannelEvent(op) + + return nil +} + +// notifyPendingOpenChannelPeerEvent updates the access manager's maps and then +// calls the channelNotifier's NotifyPendingOpenChannelEvent. +func (s *server) notifyPendingOpenChannelPeerEvent(op wire.OutPoint, + pendingChan *channeldb.OpenChannel, remotePub *btcec.PublicKey) error { + + // Call newPendingOpenChan to update the access manager's maps for this + // peer. + if err := s.peerAccessMan.newPendingOpenChan(remotePub); err != nil { + return err + } + + // Notify subscribers about this event. + s.channelNotifier.NotifyPendingOpenChannelEvent(op, pendingChan) + + return nil +} + +// notifyFundingTimeoutPeerEvent updates the access manager's maps and then +// calls the channelNotifier's NotifyFundingTimeout. +func (s *server) notifyFundingTimeoutPeerEvent(op wire.OutPoint, + remotePub *btcec.PublicKey) error { + + // Call newPendingCloseChan to potentially demote the peer. + err := s.peerAccessMan.newPendingCloseChan(remotePub) + if errors.Is(err, ErrNoMoreRestrictedAccessSlots) { + // If we encounter an error while attempting to disconnect the + // peer, log the error. + if dcErr := s.DisconnectPeer(remotePub); dcErr != nil { + srvrLog.Errorf("Unable to disconnect peer: %v\n", err) + } + } + + // Notify subscribers about this event. + s.channelNotifier.NotifyFundingTimeout(op) + + return nil +} + // peerConnected is a function that handles initialization a newly connected // peer by adding it to the server's global list of all active peers, and // starting all the goroutines the peer needs to function properly. The inbound // boolean should be true if the peer initiated the connection to us. func (s *server) peerConnected(conn net.Conn, connReq *connmgr.ConnReq, - inbound bool) { + inbound bool, access peerAccessStatus) { brontideConn := conn.(*brontide.Conn) addr := conn.RemoteAddr() @@ -4319,6 +4461,9 @@ func (s *server) peerConnected(conn net.Conn, connReq *connmgr.ConnReq, p := peer.NewBrontide(pCfg) + // Update the access manager with the access permission for this peer. + s.peerAccessMan.addPeerAccess(pubKey, access) + // TODO(roasbeef): update IP address for link-node // * also mark last-seen, do it one single transaction? @@ -4776,6 +4921,9 @@ func (s *server) removePeer(p *peer.Brontide) { delete(s.outboundPeers, pubStr) } + // Remove the peer's access permission from the access manager. + s.peerAccessMan.removePeerAccess(p.IdentityKey()) + // Copy the peer's error buffer across to the server if it has any items // in it so that we can restore peer errors across connections. if p.ErrorBuffer().Total() > 0 { diff --git a/watchtower/standalone.go b/watchtower/standalone.go index b973c9ce4..552001209 100644 --- a/watchtower/standalone.go +++ b/watchtower/standalone.go @@ -78,6 +78,7 @@ func New(cfg *Config) (*Standalone, error) { for _, listenAddr := range cfg.ListenAddrs { listener, err := brontide.NewListener( cfg.NodeKeyECDH, listenAddr.String(), + brontide.DisabledBanClosure, ) if err != nil { return nil, err From 68ec766b610cb7ddf21a8ff21f4f837808397f2b Mon Sep 17 00:00:00 2001 From: Eugene Siegel Date: Thu, 16 Jan 2025 10:43:37 -0500 Subject: [PATCH 4/7] funding+server.go: modify notifications to pass through server This modifies the various channelnotifier notification functions to instead hit the server and then call the notification routine. This allows us to accurately modify the server's maps. --- funding/manager.go | 49 ++++++++++++++++++++++++++++++++++++----- funding/manager_test.go | 18 +++++++++++++-- server.go | 5 +++-- 3 files changed, 62 insertions(+), 10 deletions(-) diff --git a/funding/manager.go b/funding/manager.go index 86f49e0eb..fb3b599d1 100644 --- a/funding/manager.go +++ b/funding/manager.go @@ -511,7 +511,7 @@ type Config struct { // NotifyOpenChannelEvent informs the ChannelNotifier when channels // transition from pending open to open. - NotifyOpenChannelEvent func(wire.OutPoint) + NotifyOpenChannelEvent func(wire.OutPoint, *btcec.PublicKey) error // OpenChannelPredicate is a predicate on the lnwire.OpenChannel message // and on the requesting node's public key that returns a bool which @@ -521,7 +521,13 @@ type Config struct { // NotifyPendingOpenChannelEvent informs the ChannelNotifier when // channels enter a pending state. NotifyPendingOpenChannelEvent func(wire.OutPoint, - *channeldb.OpenChannel) + *channeldb.OpenChannel, *btcec.PublicKey) error + + // NotifyFundingTimeout informs the ChannelNotifier when a pending-open + // channel times out because the funding transaction hasn't confirmed. + // This is only called for the fundee and only if the channel is + // zero-conf. + NotifyFundingTimeout func(wire.OutPoint, *btcec.PublicKey) error // EnableUpfrontShutdown specifies whether the upfront shutdown script // is enabled. @@ -1313,7 +1319,13 @@ func (f *Manager) advancePendingChannelState(channel *channeldb.OpenChannel, // Inform the ChannelNotifier that the channel has transitioned // from pending open to open. - f.cfg.NotifyOpenChannelEvent(channel.FundingOutpoint) + if err := f.cfg.NotifyOpenChannelEvent( + channel.FundingOutpoint, channel.IdentityPub, + ); err != nil { + log.Errorf("Unable to notify open channel event for "+ + "ChannelPoint(%v): %v", + channel.FundingOutpoint, err) + } // Find and close the discoverySignal for this channel such // that ChannelReady messages will be processed. @@ -2654,7 +2666,12 @@ func (f *Manager) fundeeProcessFundingCreated(peer lnpeer.Peer, // Inform the ChannelNotifier that the channel has entered // pending open state. - f.cfg.NotifyPendingOpenChannelEvent(fundingOut, completeChan) + if err := f.cfg.NotifyPendingOpenChannelEvent( + fundingOut, completeChan, completeChan.IdentityPub, + ); err != nil { + log.Errorf("Unable to send pending-open channel event for "+ + "ChannelPoint(%v) %v", fundingOut, err) + } // At this point we have sent our last funding message to the // initiating peer before the funding transaction will be broadcast. @@ -2874,7 +2891,14 @@ func (f *Manager) funderProcessFundingSigned(peer lnpeer.Peer, case resCtx.updates <- upd: // Inform the ChannelNotifier that the channel has entered // pending open state. - f.cfg.NotifyPendingOpenChannelEvent(*fundingPoint, completeChan) + if err := f.cfg.NotifyPendingOpenChannelEvent( + *fundingPoint, completeChan, completeChan.IdentityPub, + ); err != nil { + log.Errorf("Unable to send pending-open channel "+ + "event for ChannelPoint(%v) %v", fundingPoint, + err) + } + case <-f.quit: return } @@ -2930,6 +2954,13 @@ func (f *Manager) fundingTimeout(c *channeldb.OpenChannel, c.FundingOutpoint, err) } + // Notify other subsystems about the funding timeout. + err := f.cfg.NotifyFundingTimeout(c.FundingOutpoint, c.IdentityPub) + if err != nil { + log.Errorf("failed to notify of funding timeout for "+ + "ChanPoint(%v): %v", c.FundingOutpoint, err) + } + timeoutErr := fmt.Errorf("timeout waiting for funding tx (%v) to "+ "confirm", c.FundingOutpoint) @@ -3310,7 +3341,13 @@ func (f *Manager) handleFundingConfirmation( // Inform the ChannelNotifier that the channel has transitioned from // pending open to open. - f.cfg.NotifyOpenChannelEvent(completeChan.FundingOutpoint) + if err := f.cfg.NotifyOpenChannelEvent( + completeChan.FundingOutpoint, completeChan.IdentityPub, + ); err != nil { + log.Errorf("Unable to notify open channel event for "+ + "ChannelPoint(%v): %v", completeChan.FundingOutpoint, + err) + } // Close the discoverySignal channel, indicating to a separate // goroutine that the channel now is marked as open in the database diff --git a/funding/manager_test.go b/funding/manager_test.go index ad4a11b49..6a1b35ed2 100644 --- a/funding/manager_test.go +++ b/funding/manager_test.go @@ -231,17 +231,30 @@ type mockChanEvent struct { pendingOpenEvent chan channelnotifier.PendingOpenChannelEvent } -func (m *mockChanEvent) NotifyOpenChannelEvent(outpoint wire.OutPoint) { +func (m *mockChanEvent) NotifyOpenChannelEvent(outpoint wire.OutPoint, + remotePub *btcec.PublicKey) error { + m.openEvent <- outpoint + + return nil } func (m *mockChanEvent) NotifyPendingOpenChannelEvent(outpoint wire.OutPoint, - pendingChannel *channeldb.OpenChannel) { + pendingChannel *channeldb.OpenChannel, + remotePub *btcec.PublicKey) error { m.pendingOpenEvent <- channelnotifier.PendingOpenChannelEvent{ ChannelPoint: &outpoint, PendingChannel: pendingChannel, } + + return nil +} + +func (m *mockChanEvent) NotifyFundingTimeout(outpoint wire.OutPoint, + remotePub *btcec.PublicKey) error { + + return nil } // mockZeroConfAcceptor always accepts the channel open request for zero-conf @@ -550,6 +563,7 @@ func createTestFundingManager(t *testing.T, privKey *btcec.PrivateKey, NotifyOpenChannelEvent: evt.NotifyOpenChannelEvent, OpenChannelPredicate: chainedAcceptor, NotifyPendingOpenChannelEvent: evt.NotifyPendingOpenChannelEvent, + NotifyFundingTimeout: evt.NotifyFundingTimeout, DeleteAliasEdge: func(scid lnwire.ShortChannelID) ( *models.ChannelEdgePolicy, error) { diff --git a/server.go b/server.go index 65672c9e1..a848340d5 100644 --- a/server.go +++ b/server.go @@ -1681,9 +1681,10 @@ func newServer(cfg *Config, listenAddrs []net.Addr, MaxPendingChannels: cfg.MaxPendingChannels, RejectPush: cfg.RejectPush, MaxLocalCSVDelay: chainCfg.MaxLocalDelay, - NotifyOpenChannelEvent: s.channelNotifier.NotifyOpenChannelEvent, + NotifyOpenChannelEvent: s.notifyOpenChannelPeerEvent, OpenChannelPredicate: chanPredicate, - NotifyPendingOpenChannelEvent: s.channelNotifier.NotifyPendingOpenChannelEvent, + NotifyPendingOpenChannelEvent: s.notifyPendingOpenChannelPeerEvent, + NotifyFundingTimeout: s.notifyFundingTimeoutPeerEvent, EnableUpfrontShutdown: cfg.EnableUpfrontShutdown, MaxAnchorsCommitFeeRate: chainfee.SatPerKVByte( s.cfg.MaxCommitFeeRateAnchors * 1000).FeePerKWeight(), From d5ecad3834c64cfed29a023a8e37626d7af05409 Mon Sep 17 00:00:00 2001 From: Eugene Siegel Date: Wed, 12 Feb 2025 11:50:14 -0500 Subject: [PATCH 5/7] itest: new test to check server access perms --- itest/list_on_test.go | 4 ++ itest/lnd_access_perm_test.go | 93 +++++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+) create mode 100644 itest/lnd_access_perm_test.go diff --git a/itest/list_on_test.go b/itest/list_on_test.go index 78e242b1a..e1160ed0f 100644 --- a/itest/list_on_test.go +++ b/itest/list_on_test.go @@ -674,6 +674,10 @@ var allTestCases = []*lntest.TestCase{ Name: "funding manager funding timeout", TestFunc: testFundingManagerFundingTimeout, }, + { + Name: "access perm", + TestFunc: testAccessPerm, + }, } // appendPrefixed is used to add a prefix to each test name in the subtests diff --git a/itest/lnd_access_perm_test.go b/itest/lnd_access_perm_test.go new file mode 100644 index 000000000..6ff0fbb24 --- /dev/null +++ b/itest/lnd_access_perm_test.go @@ -0,0 +1,93 @@ +package itest + +import ( + "strconv" + + "github.com/lightningnetwork/lnd/lnrpc" + "github.com/lightningnetwork/lnd/lntest" +) + +// testAccessPerm tests that the number of restricted slots is upheld when +// connecting to the server from a restrictedd peer. +func testAccessPerm(ht *lntest.HarnessTest) { + args := []string{ + "--minbackoff=5m", + "--maxbackoff=5m", + "--num-restricted-slots=5", + } + + alice := ht.NewNodeWithCoins("Alice", args) + bob := ht.NewNodeWithCoins("Bob", args) + ht.ConnectNodes(alice, bob) + + // Open a confirmed channel to Bob. Bob will have protected access. + chanPoint1 := ht.OpenChannel( + alice, bob, lntest.OpenChannelParams{ + Amt: chanAmt, + }, + ) + defer ht.CloseChannel(alice, chanPoint1) + + // Open and close channel to Carol. Carol will have protected access. + carol := ht.NewNodeWithCoins("Carol", args) + ht.ConnectNodes(alice, carol) + + chanPoint2 := ht.OpenChannel( + alice, carol, lntest.OpenChannelParams{ + Amt: chanAmt, + }, + ) + + ht.CloseChannel(alice, chanPoint2) + + // Make a pending channel with Dave. + dave := ht.NewNodeWithCoins("Dave", args) + ht.ConnectNodes(alice, dave) + + ht.OpenChannelAssertStream( + dave, alice, lntest.OpenChannelParams{ + Amt: chanAmt, + }, + ) + + // Disconnect Bob, Carol, and Dave. + ht.DisconnectNodes(alice, bob) + ht.AssertNotConnected(alice, bob) + + ht.DisconnectNodes(alice, carol) + ht.AssertNotConnected(alice, carol) + + ht.DisconnectNodes(alice, dave) + ht.AssertNotConnected(alice, dave) + + // Connect 5 times to Alice. All of these connections should be + // successful. + for i := 0; i < 5; i++ { + peer := ht.NewNode("Peer"+strconv.Itoa(i), args) + ht.ConnectNodes(peer, alice) + ht.AssertConnected(peer, alice) + } + + // Connect an additional time to Alice. This should fail. + failedPeer := ht.NewNode("FailedPeer", args) + req := &lnrpc.ConnectPeerRequest{ + Addr: &lnrpc.LightningAddress{ + Pubkey: alice.RPC.GetInfo().IdentityPubkey, + Host: alice.Cfg.P2PAddr(), + }, + } + _ = failedPeer.RPC.ConnectPeer(req) + ht.AssertNotConnected(failedPeer, alice) + + // Connect nodes and assert access status. + ht.ConnectNodes(alice, bob) + ht.AssertConnected(alice, bob) + + ht.ConnectNodes(alice, carol) + ht.AssertConnected(alice, carol) + + ht.ConnectNodes(alice, dave) + ht.AssertConnected(alice, dave) + + ht.MineBlocksAndAssertNumTxes(1, 1) +} From a4acfcb0ef5d7660f5f382943a410beae28e083c Mon Sep 17 00:00:00 2001 From: Eugene Siegel Date: Thu, 13 Feb 2025 13:19:26 -0500 Subject: [PATCH 6/7] sample-lnd.conf: update for num-restricted-slots --- sample-lnd.conf | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sample-lnd.conf b/sample-lnd.conf index b447e6b6f..f3123a0d7 100644 --- a/sample-lnd.conf +++ b/sample-lnd.conf @@ -566,6 +566,8 @@ ; the headers of an HTTP request. ; http-header-timeout=5s +; The number of restricted slots the server will allocate for peers. +; num-restricted-slots=30 [fee] From 6309b8a0f4667494bc00dd4ad159e1ad59de6a61 Mon Sep 17 00:00:00 2001 From: Eugene Siegel Date: Thu, 13 Feb 2025 13:32:54 -0500 Subject: [PATCH 7/7] release-notes: update for 0.19.0 --- docs/release-notes/release-notes-0.19.0.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/docs/release-notes/release-notes-0.19.0.md b/docs/release-notes/release-notes-0.19.0.md index f0a400708..74eabc684 100644 --- a/docs/release-notes/release-notes-0.19.0.md +++ b/docs/release-notes/release-notes-0.19.0.md @@ -195,6 +195,9 @@ * When deciding whether `lnd` is synced to chain, the current height from the blockbeat dispatcher is now also [taken into consideration](https://github.com/lightningnetwork/lnd/pull/9501). + +* [The server](https://github.com/lightningnetwork/lnd/pull/9458) now allocates + restricted slots for certain peers. This is configured by --num-restricted-slots. ## RPC Updates