htlcswitch: bound peer-controlled channel ingress

In this commit, we bound the channel mailbox by message count and by the
encoded size of non-commitment control messages. Commitment updates retain
their full custom-record allowance and remain protected by the count bound.
If either budget fills, we disconnect the peer instead of silently dropping
an ordered channel message.

We also reject unauthorized fee updates before fee-exposure evaluation,
return the exposure error used to fail the link, and emit peer-controlled
warning classes only once per link lifetime.

htlcswitch/link.go

@@ -363,6 +363,14 @@ type channelLink struct {
 	// forwarded sent by the switch.
 	mailBox MailBox
 
+	// mailBoxIngressMtx guards mailBoxIngressFailed and serializes peer
+	// message admission into the mailbox.
+	mailBoxIngressMtx sync.Mutex
+
+	// mailBoxIngressFailed is set after the first peer message admission
+	// failure so later messages cannot be processed across a gap.
+	mailBoxIngressFailed bool
+
 	// upstream is a channel that new messages sent from the remote peer to
 	// the local peer will be sent across.
 	upstream chan lnwire.Message
@@ -395,6 +403,11 @@ type channelLink struct {
 	// log is a link-specific logging instance.
 	log btclog.Logger
 
+	// warningLogged and unknownMessageLogged track whether each non-fatal
+	// message class has already been recorded for this link lifetime.
+	warningLogged        bool
+	unknownMessageLogged bool
+
 	// isOutgoingAddBlocked tracks whether the channelLink can send an
 	// UpdateAddHTLC.
 	isOutgoingAddBlocked atomic.Bool
@@ -1862,14 +1875,20 @@ func (l *channelLink) handleUpstreamMsg(ctx context.Context,
 	// log it and move on. We choose not to disconnect from our peer,
 	// although we "MAY" do so according to the specification.
 	case *lnwire.Warning:
-		l.log.Warnf("received warning message from peer: %v",
-			msg.Warning())
+		if !l.warningLogged {
+			l.log.Warnf("received warning message from peer: %v",
+				msg.Warning())
+			l.warningLogged = true
+		}
 
 	case *lnwire.Error:
 		l.processRemoteError(msg)
 
 	default:
-		l.log.Warnf("received unknown message of type %T", msg)
+		if !l.unknownMessageLogged {
+			l.log.Warnf("received unknown message of type %T", msg)
+			l.unknownMessageLogged = true
+		}
 	}
 
 	if err != nil {
@@ -2804,10 +2823,23 @@ func (l *channelLink) HandleChannelUpdate(message lnwire.Message) {
 	default:
 	}
 
-	err := l.mailBox.AddMessage(message)
-	if err != nil {
-		l.log.Errorf("failed to add Message to mailbox: %v", err)
+	l.mailBoxIngressMtx.Lock()
+	if l.mailBoxIngressFailed {
+		l.mailBoxIngressMtx.Unlock()
+		return
 	}
+
+	err := l.mailBox.AddMessage(message)
+	if err == nil {
+		l.mailBoxIngressMtx.Unlock()
+		return
+	}
+
+	l.mailBoxIngressFailed = true
+	l.mailBoxIngressMtx.Unlock()
+
+	l.log.Errorf("failed to add Message to mailbox: %v", err)
+	go l.cfg.Peer.Disconnect(err)
 }
 
 // updateChannelFee updates the commitment fee-per-kw on this channel by
@@ -4583,6 +4615,16 @@ func (l *channelLink) processRemoteRevokeAndAck(ctx context.Context,
 // processRemoteUpdateFee takes an `UpdateFee` msg sent from the remote and
 // processes it.
 func (l *channelLink) processRemoteUpdateFee(msg *lnwire.UpdateFee) error {
+	// BOLT 2 only permits the channel initiator to send fee updates.
+	// Validate the sender's role before applying message-specific
+	// calculations.
+	if l.channel.IsInitiator() {
+		err := fmt.Errorf("received fee update as initiator")
+		l.failf(LinkFailureError{code: ErrInvalidUpdate}, "%v", err)
+
+		return err
+	}
+
 	// Check and see if their proposed fee-rate would make us exceed the fee
 	// threshold.
 	fee := chainfee.SatPerKWeight(msg.FeePerKw)
@@ -4601,8 +4643,9 @@ func (l *channelLink) processRemoteUpdateFee(msg *lnwire.UpdateFee) error {
 
 	if isDust {
 		// The proposed fee-rate makes us exceed the fee threshold.
-		l.failf(LinkFailureError{code: ErrInternalError},
-			"fee threshold exceeded: %v", err)
+		err := fmt.Errorf("fee threshold exceeded")
+		l.failf(LinkFailureError{code: ErrInternalError}, "%v", err)
+
 		return err
 	}
 
@@ -4611,6 +4654,7 @@ func (l *channelLink) processRemoteUpdateFee(msg *lnwire.UpdateFee) error {
 	if err := l.channel.ReceiveUpdateFee(fee); err != nil {
 		l.failf(LinkFailureError{code: ErrInvalidUpdate},
 			"error receiving fee update: %v", err)
+
 		return err
 	}
 

htlcswitch/link_fee_update_test.go

@@ -0,0 +1,306 @@
+package htlcswitch
+
+import (
+	"bytes"
+	"errors"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/lnpeer"
+	"github.com/lightningnetwork/lnd/lnwallet"
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/stretchr/testify/require"
+)
+
+// mailboxAdmissionPeer records disconnect requests made by a channel link.
+type mailboxAdmissionPeer struct {
+	*lnpeer.MockPeer
+
+	disconnected chan error
+}
+
+// Disconnect records the error supplied by the channel link.
+func (p *mailboxAdmissionPeer) Disconnect(err error) {
+	p.disconnected <- err
+}
+
+// mailboxAdmissionTestBox fails its first message admission and records the
+// number of admission attempts.
+type mailboxAdmissionTestBox struct {
+	MailBox
+
+	mu       sync.Mutex
+	addCalls int
+}
+
+// AddMessage records an admission attempt and fails the first one.
+func (m *mailboxAdmissionTestBox) AddMessage(lnwire.Message) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.addCalls++
+	if m.addCalls == 1 {
+		return errWireMessageQueueFull
+	}
+
+	return nil
+}
+
+// calls returns the number of message admission attempts.
+func (m *mailboxAdmissionTestBox) calls() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	return m.addCalls
+}
+
+// newLinkCapturingLogger returns a logger backed by an in-memory buffer.
+func newLinkCapturingLogger() (btclog.Logger, *bytes.Buffer) {
+	buf := &bytes.Buffer{}
+	handler := btclog.NewDefaultHandler(buf, btclog.WithNoTimestamp())
+
+	return btclog.NewSLogger(handler), buf
+}
+
+// TestProcessRemoteUpdateFeeRoleValidation checks that fee update role
+// validation is performed at the link boundary.
+func TestProcessRemoteUpdateFeeRoleValidation(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, bobChannel, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	newLink := func(channel *lnwallet.LightningChannel) *channelLink {
+		link, ok := NewChannelLink(ChannelLinkConfig{
+			DisallowQuiescence: true,
+			OnChannelFailure: func(lnwire.ChannelID,
+				lnwire.ShortChannelID, LinkFailureError) {
+			},
+		}, channel).(*channelLink)
+		require.True(t, ok)
+
+		return link
+	}
+
+	t.Run("unauthorized sender", func(t *testing.T) {
+		link := newLink(aliceChannel)
+
+		err := link.processRemoteUpdateFee(&lnwire.UpdateFee{})
+		require.EqualError(t, err, "received fee update as initiator")
+		require.True(t, link.failed)
+	})
+
+	t.Run("authorized sender", func(t *testing.T) {
+		link := newLink(bobChannel)
+		mailbox := newMemoryMailBox(&mailBoxConfig{})
+		link.mailBox = mailbox
+
+		feeRate := bobChannel.CommitFeeRate() + 1
+		err := link.processRemoteUpdateFee(&lnwire.UpdateFee{
+			FeePerKw: uint32(feeRate),
+		})
+		require.NoError(t, err)
+		require.False(t, link.failed)
+		require.True(t, bobChannel.NeedCommitment())
+		require.Equal(t, feeRate, mailbox.feeRate)
+	})
+}
+
+// TestProcessRemoteUpdateFeeExposureError checks that exceeding the fee
+// exposure limit returns the error used to fail the link.
+func TestProcessRemoteUpdateFeeExposureError(t *testing.T) {
+	t.Parallel()
+
+	_, bobChannel, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		DisallowQuiescence: true,
+		MaxFeeExposure:     1,
+		OnChannelFailure: func(lnwire.ChannelID,
+			lnwire.ShortChannelID, LinkFailureError) {
+		},
+	}, bobChannel).(*channelLink)
+	require.True(t, ok)
+
+	err = link.processRemoteUpdateFee(&lnwire.UpdateFee{
+		FeePerKw: 1000,
+	})
+	require.EqualError(t, err, "fee threshold exceeded")
+	require.True(t, link.failed)
+}
+
+// TestLinkLogDeduplication checks that repeated non-fatal message classes are
+// only recorded once during a link lifetime.
+func TestLinkLogDeduplication(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, _, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		DisallowQuiescence: true,
+	}, aliceChannel).(*channelLink)
+	require.True(t, ok)
+	logger, logBuffer := newLinkCapturingLogger()
+	link.log = logger
+
+	for i := 0; i < 2; i++ {
+		link.handleUpstreamMsg(t.Context(), &lnwire.Warning{})
+		link.handleUpstreamMsg(
+			t.Context(), &lnwire.ChannelReestablish{},
+		)
+	}
+
+	warningCount := strings.Count(
+		logBuffer.String(), "received warning message from peer",
+	)
+	require.Equal(t, 1, warningCount)
+	require.Equal(
+		t, 1, strings.Count(
+			logBuffer.String(), "received unknown message of type",
+		),
+	)
+}
+
+// TestChannelMessageAdmissionError checks that an admission error reconnects
+// the ordered channel message stream instead of omitting a message.
+func TestChannelMessageAdmissionError(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, _, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	peer := &mailboxAdmissionPeer{
+		MockPeer:     &lnpeer.MockPeer{},
+		disconnected: make(chan error, 1),
+	}
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		Peer:               peer,
+		DisallowQuiescence: true,
+	}, aliceChannel).(*channelLink)
+	require.True(t, ok)
+
+	mailbox := newMemoryMailBox(&mailBoxConfig{})
+	link.mailBox = mailbox
+	for i := 0; i < maxWireMessages; i++ {
+		require.NoError(t, mailbox.AddMessage(&lnwire.UpdateFee{}))
+	}
+
+	link.HandleChannelUpdate(&lnwire.UpdateFee{})
+
+	select {
+	case err := <-peer.disconnected:
+		require.ErrorIs(t, err, errWireMessageQueueFull)
+
+	case <-time.After(time.Second):
+		t.Fatal("mailbox admission error did not disconnect peer")
+	}
+}
+
+// TestChannelMessageAdmissionFailureLatch checks that a link stops admitting
+// peer messages after its first mailbox admission failure.
+func TestChannelMessageAdmissionFailureLatch(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, _, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	peer := &mailboxAdmissionPeer{
+		MockPeer:     &lnpeer.MockPeer{},
+		disconnected: make(chan error, 2),
+	}
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		Peer:               peer,
+		DisallowQuiescence: true,
+	}, aliceChannel).(*channelLink)
+	require.True(t, ok)
+
+	mailbox := &mailboxAdmissionTestBox{}
+	link.mailBox = mailbox
+	logger, logBuffer := newLinkCapturingLogger()
+	link.log = logger
+
+	link.HandleChannelUpdate(&lnwire.UpdateFee{})
+
+	select {
+	case err := <-peer.disconnected:
+		require.ErrorIs(t, err, errWireMessageQueueFull)
+
+	case <-time.After(time.Second):
+		t.Fatal("mailbox admission error did not disconnect peer")
+	}
+
+	link.HandleChannelUpdate(&lnwire.CommitSig{})
+
+	require.Equal(t, 1, mailbox.calls())
+	require.Equal(
+		t, 1, strings.Count(
+			logBuffer.String(), "failed to add Message to mailbox",
+		),
+	)
+	select {
+	case err := <-peer.disconnected:
+		t.Fatalf("unexpected second disconnect: %v", err)
+
+	default:
+	}
+}
+
+// TestChannelMessageSizeAdmissionError checks that a message-size admission
+// error reconnects the ordered channel message stream.
+func TestChannelMessageSizeAdmissionError(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, _, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	peer := &mailboxAdmissionPeer{
+		MockPeer:     &lnpeer.MockPeer{},
+		disconnected: make(chan error, 1),
+	}
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		Peer:               peer,
+		DisallowQuiescence: true,
+	}, aliceChannel).(*channelLink)
+	require.True(t, ok)
+
+	mailbox := newMemoryMailBox(&mailBoxConfig{})
+	link.mailBox = mailbox
+	msg := &lnwire.Warning{
+		Data: make([]byte, lnwire.MaxMsgBody-40),
+	}
+	for {
+		err := mailbox.AddMessage(msg)
+		if errors.Is(err, errWireMessageQueueFull) {
+			break
+		}
+		require.NoError(t, err)
+	}
+
+	link.HandleChannelUpdate(msg)
+
+	select {
+	case err := <-peer.disconnected:
+		require.ErrorIs(t, err, errWireMessageQueueFull)
+
+	case <-time.After(time.Second):
+		t.Fatal("message-size admission error did not disconnect peer")
+	}
+}

htlcswitch/mailbox.go

@@ -14,6 +14,16 @@ import (
 	"github.com/lightningnetwork/lnd/lnwire"
 )
 
+const (
+	// maxWireMessages is the maximum number of ordered messages that can
+	// wait for a channel link. It accommodates a full commitment batch.
+	maxWireMessages = 1000
+
+	// maxWireBytes bounds the encoded size of messages that can wait for a
+	// channel link.
+	maxWireBytes = 4 * 1024 * 1024
+)
+
 var (
 	// ErrMailBoxShuttingDown is returned when the mailbox is interrupted by
 	// a shutdown request.
@@ -22,6 +32,12 @@ var (
 	// ErrPacketAlreadyExists signals that an attempt to add a packet failed
 	// because it already exists in the mailbox.
 	ErrPacketAlreadyExists = errors.New("mailbox already has packet")
+
+	// errWireMessageQueueFull signals that the wire-message queue has
+	// reached one of its admission budgets.
+	errWireMessageQueueFull = errors.New(
+		"mailbox wire message queue is full",
+	)
 )
 
 // MailBox is an interface which represents a concurrent-safe, in-order
@@ -122,6 +138,7 @@ type memoryMailBox struct {
 	cfg *mailBoxConfig
 
 	wireMessages *list.List
+	wireBytes    uint32
 	wireMtx      sync.Mutex
 	wireCond     *sync.Cond
 
@@ -160,6 +177,13 @@ type memoryMailBox struct {
 	isDust dustClosure
 }
 
+// queuedWireMessage stores a wire message and its encoded size charged to the
+// wire-message budget.
+type queuedWireMessage struct {
+	msg  lnwire.Message
+	size uint32
+}
+
 // newMemoryMailBox creates a new instance of the memoryMailBox.
 func newMemoryMailBox(cfg *mailBoxConfig) *memoryMailBox {
 	box := &memoryMailBox{
@@ -383,6 +407,7 @@ func (m *memoryMailBox) wireMailCourier() {
 			select {
 			case msgDone := <-m.msgReset:
 				m.wireMessages.Init()
+				m.wireBytes = 0
 				close(msgDone)
 			case <-m.quit:
 				m.wireCond.L.Unlock()
@@ -397,7 +422,9 @@ func (m *memoryMailBox) wireMailCourier() {
 		entry := m.wireMessages.Front()
 
 		//nolint:forcetypeassert
-		nextMsg := m.wireMessages.Remove(entry).(lnwire.Message)
+		queuedMsg := m.wireMessages.Remove(entry).(*queuedWireMessage)
+		m.wireBytes -= queuedMsg.size
+		nextMsg := queuedMsg.msg
 
 		// Now that we're done with the condition, we can unlock it to
 		// allow any callers to append to the end of our target queue.
@@ -411,6 +438,7 @@ func (m *memoryMailBox) wireMailCourier() {
 		case msgDone := <-m.msgReset:
 			m.wireCond.L.Lock()
 			m.wireMessages.Init()
+			m.wireBytes = 0
 			m.wireCond.L.Unlock()
 
 			close(msgDone)
@@ -560,10 +588,28 @@ func (m *memoryMailBox) pktMailCourier() {
 // NOTE: This method is safe for concrete use and part of the MailBox
 // interface.
 func (m *memoryMailBox) AddMessage(msg lnwire.Message) error {
+	msgSize, err := wireMessageSize(msg)
+	if err != nil {
+		return fmt.Errorf(
+			"unable to determine wire message size: %w", err,
+		)
+	}
+
 	// First, we'll lock the condition, and add the message to the end of
 	// the wire message inbox.
 	m.wireCond.L.Lock()
-	m.wireMessages.PushBack(msg)
+	if m.wireMessages.Len() >= maxWireMessages ||
+		m.wireBytes+msgSize > maxWireBytes {
+
+		m.wireCond.L.Unlock()
+		return errWireMessageQueueFull
+	}
+
+	m.wireMessages.PushBack(&queuedWireMessage{
+		msg:  msg,
+		size: msgSize,
+	})
+	m.wireBytes += msgSize
 	m.wireCond.L.Unlock()
 
 	// With the message added, we signal to the mailCourier that there are
@@ -573,6 +619,16 @@ func (m *memoryMailBox) AddMessage(msg lnwire.Message) error {
 	return nil
 }
 
+// wireMessageSize returns the serialized bytes charged to the wire-message
+// budget.
+func wireMessageSize(msg lnwire.Message) (uint32, error) {
+	if sizeableMsg, ok := msg.(lnwire.SizeableMessage); ok {
+		return sizeableMsg.SerializedSize()
+	}
+
+	return lnwire.MessageSerializedSize(msg)
+}
+
 // AddPacket appends a new message to the end of the packet queue.
 //
 // NOTE: This method is safe for concrete use and part of the MailBox

htlcswitch/mailbox_test.go

@@ -1,11 +1,13 @@
 package htlcswitch
 
 import (
+	"errors"
 	prand "math/rand"
 	"reflect"
 	"testing"
 	"time"
 
+	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/btcutil/v2"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/lightningnetwork/lnd/channeldb"
@@ -168,6 +170,96 @@ func TestMailBoxCouriers(t *testing.T) {
 	}
 }
 
+// TestMailBoxAdmissionBudgets checks message-count and serialized-size
+// admission behavior for the wire-message queue.
+func TestMailBoxAdmissionBudgets(t *testing.T) {
+	t.Parallel()
+
+	t.Run("message count", func(t *testing.T) {
+		mailbox := newMemoryMailBox(&mailBoxConfig{})
+		msg := &lnwire.UpdateFee{}
+
+		for i := 0; i < maxWireMessages; i++ {
+			require.NoError(t, mailbox.AddMessage(msg))
+		}
+
+		require.ErrorIs(
+			t, mailbox.AddMessage(msg), errWireMessageQueueFull,
+		)
+		require.Equal(t, maxWireMessages, mailbox.wireMessages.Len())
+		require.LessOrEqual(
+			t, mailbox.wireBytes, uint32(maxWireBytes),
+		)
+	})
+
+	t.Run("encoded bytes", func(t *testing.T) {
+		mailbox := newMemoryMailBox(&mailBoxConfig{})
+		msg := &lnwire.Warning{
+			Data: make([]byte, lnwire.MaxMsgBody-40),
+		}
+
+		for {
+			err := mailbox.AddMessage(msg)
+			if errors.Is(err, errWireMessageQueueFull) {
+				break
+			}
+			require.NoError(t, err)
+		}
+
+		require.Less(t, mailbox.wireMessages.Len(), maxWireMessages)
+		require.LessOrEqual(
+			t, mailbox.wireBytes, uint32(maxWireBytes),
+		)
+	})
+
+	t.Run("commitment message sizes", func(t *testing.T) {
+		_, pubKey := btcec.PrivKeyFromBytes(make([]byte, 32))
+		extraData := lnwire.ExtraOpaqueData{
+			0xfe, 0x00, 0x01, 0x00, 0x00, 0x03, 0x01, 0x02, 0x03,
+		}
+
+		messages := []lnwire.Message{
+			&lnwire.CommitSig{ExtraData: extraData},
+			&lnwire.RevokeAndAck{
+				NextRevocationKey: pubKey,
+				ExtraData:         extraData,
+			},
+			&lnwire.Stfu{ExtraData: extraData},
+		}
+		for _, msg := range messages {
+			mailbox := newMemoryMailBox(&mailBoxConfig{})
+			sizeableMsg, ok := msg.(lnwire.SizeableMessage)
+			require.True(t, ok)
+
+			expectedSize, err := sizeableMsg.SerializedSize()
+			require.NoError(t, err)
+
+			require.NoError(t, mailbox.AddMessage(msg))
+			require.Equal(t, expectedSize, mailbox.wireBytes)
+		}
+	})
+
+	t.Run("reset restores byte budget", func(t *testing.T) {
+		mailbox := newMemoryMailBox(&mailBoxConfig{})
+		mailbox.Start()
+		t.Cleanup(mailbox.Stop)
+
+		msg := &lnwire.Warning{
+			Data: make([]byte, lnwire.MaxMsgBody-40),
+		}
+		for {
+			err := mailbox.AddMessage(msg)
+			if errors.Is(err, errWireMessageQueueFull) {
+				break
+			}
+			require.NoError(t, err)
+		}
+
+		require.NoError(t, mailbox.ResetMessages())
+		require.NoError(t, mailbox.AddMessage(msg))
+	})
+}
+
 // TestMailBoxResetAfterShutdown tests that ResetMessages and ResetPackets
 // return ErrMailBoxShuttingDown after the mailbox has been stopped.
 func TestMailBoxResetAfterShutdown(t *testing.T) {