Merge pull request #9458 from Crypt-iQ/banning_010072025

multi+server.go: add initial permissions for some peers

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)
+}

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)
+}

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
+}

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
 	}

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
 }
 

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

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()
 

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) {

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,
 	}
 }
 

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{}),
 	}

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
 

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 (
@@ -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

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) {
 

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

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)
+}

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,
 	)
 }

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]
 

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{
@@ -1625,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(),
@@ -1816,6 +1873,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 +2471,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 +3701,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 +3971,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 +4020,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 +4052,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 +4077,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 +4154,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 +4188,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 +4262,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 +4462,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 +4922,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 {

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