multi: endpoints for onion messages

This commit creates the necessary endpoints for onion messages.
Specifically, it adds the following:

- `SendOnionMessage` endpoint to send onion messages.
- `SubscribeOnionMessages` endpoint to subscribe to incoming onion
  messages.

It uses the `msgmux` package to handle the onion messages.

Author: gijswijs

itest/list_on_test.go

@@ -531,6 +531,10 @@ var allTestCases = []*lntest.TestCase{
 		Name:     "custom message",
 		TestFunc: testCustomMessage,
 	},
+	{
+		Name:     "onion message",
+		TestFunc: testOnionMessage,
+	},
 	{
 		Name:     "sign verify message with addr",
 		TestFunc: testSignVerifyMessageWithAddr,

itest/lnd_onion_message_test.go

@@ -0,0 +1,75 @@
+package itest
+
+import (
+	"time"
+
+	"github.com/btcsuite/btcd/btcec/v2"
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"github.com/lightningnetwork/lnd/lntest"
+	"github.com/stretchr/testify/require"
+)
+
+// testOnionMessage tests sending and receiving of the onion message type.
+func testOnionMessage(ht *lntest.HarnessTest) {
+	alice := ht.NewNode("Alice", nil)
+	bob := ht.NewNode("Bob", nil)
+
+	// Subscribe Alice to onion messages before we send any, so that we
+	// don't miss any.
+	msgClient, cancel := alice.RPC.SubscribeOnionMessages()
+	defer cancel()
+
+	// Create a channel to receive onion messages on.
+	messages := make(chan *lnrpc.OnionMessage)
+	go func() {
+		for {
+			// If we fail to receive, just exit. The test should
+			// fail elsewhere if it doesn't get a message that it
+			// was expecting.
+			msg, err := msgClient.Recv()
+			if err != nil {
+				return
+			}
+
+			// Deliver the message into our channel or exit if the
+			// test is shutting down.
+			select {
+			case messages <- msg:
+			case <-ht.Context().Done():
+				return
+			}
+		}
+	}()
+
+	// Connect alice and bob so that they can exchange messages.
+	ht.EnsureConnected(alice, bob)
+
+	// Create a random onion message.
+	randomPriv, err := btcec.NewPrivateKey()
+	require.NoError(ht.T, err)
+	randomPub := randomPriv.PubKey()
+	msgPathKey := randomPub.SerializeCompressed()
+	msgOnion := []byte{1, 2, 3}
+
+	// Send it from Bob to Alice.
+	bobMsg := &lnrpc.SendOnionMessageRequest{
+		Peer:    alice.PubKey[:],
+		PathKey: msgPathKey,
+		Onion:   msgOnion,
+	}
+	bob.RPC.SendOnionMessage(bobMsg)
+
+	// Wait for Alice to receive the message.
+	select {
+	case msg := <-messages:
+		// Check our type and data and (sanity) check the peer we got
+		// it from.
+		require.Equal(ht, msgOnion, msg.Onion, "msg data wrong")
+		require.Equal(ht, msgPathKey, msg.PathKey, "msg "+
+			"path key wrong")
+		require.Equal(ht, bob.PubKey[:], msg.Peer, "msg peer wrong")
+
+	case <-time.After(lntest.DefaultTimeout):
+		ht.Fatalf("alice did not receive onion message: %v", bobMsg)
+	}
+}

lntest/rpc/lnd.go

@@ -726,6 +726,8 @@ func (h *HarnessRPC) SubscribeChannelEvents() ChannelEventsClient {
 
 type CustomMessageClient lnrpc.Lightning_SubscribeCustomMessagesClient
 
+type OnionMessageClient lnrpc.Lightning_SubscribeOnionMessagesClient
+
 // SubscribeCustomMessages creates a subscription client for custom messages.
 func (h *HarnessRPC) SubscribeCustomMessages() (CustomMessageClient,
 	context.CancelFunc) {
@@ -758,6 +760,38 @@ func (h *HarnessRPC) SendCustomMessage(
 	return resp
 }
 
+// SendOnionMessage makes a RPC call to the node's SendOnionMessage and
+// returns the response.
+func (h *HarnessRPC) SendOnionMessage(
+	req *lnrpc.SendOnionMessageRequest) *lnrpc.SendOnionMessageResponse {
+
+	ctxt, cancel := context.WithTimeout(h.runCtx, DefaultTimeout)
+	defer cancel()
+
+	resp, err := h.LN.SendOnionMessage(ctxt, req)
+	h.NoError(err, "SendOnionMessage")
+
+	return resp
+}
+
+// SubscribeOnionMessages creates a subscription client for onion messages.
+func (h *HarnessRPC) SubscribeOnionMessages() (OnionMessageClient,
+	context.CancelFunc) {
+
+	ctxt, cancel := context.WithCancel(h.runCtx)
+
+	req := &lnrpc.SubscribeOnionMessagesRequest{}
+
+	// SubscribeCustomMessages needs to have the context alive for the
+	// entire test case as the returned client will be used for send and
+	// receive events stream. Thus we use runCtx here instead of a timeout
+	// context.
+	stream, err := h.LN.SubscribeOnionMessages(ctxt, req)
+	h.NoError(err, "SubscribeOnionMessages")
+
+	return stream, cancel
+}
+
 // GetChanInfo makes a RPC call to the node's GetChanInfo and returns the
 // response.
 func (h *HarnessRPC) GetChanInfo(

log.go

@@ -46,6 +46,7 @@ import (
 	"github.com/lightningnetwork/lnd/monitoring"
 	"github.com/lightningnetwork/lnd/msgmux"
 	"github.com/lightningnetwork/lnd/netann"
+	"github.com/lightningnetwork/lnd/onionmessage"
 	paymentsdb "github.com/lightningnetwork/lnd/payments/db"
 	"github.com/lightningnetwork/lnd/peer"
 	"github.com/lightningnetwork/lnd/peernotifier"
@@ -212,6 +213,7 @@ func SetupLoggers(root *build.SubLoggerManager, interceptor signal.Interceptor)
 		root, paymentsdb.Subsystem, interceptor, paymentsdb.UseLogger,
 	)
 
+	AddSubLogger(root, onionmessage.Subsystem, interceptor, onionmessage.UseLogger)
 }
 
 // AddSubLogger is a helper method to conveniently create and register the

onionmessage/log.go

@@ -0,0 +1,32 @@
+package onionmessage
+
+import (
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+// Subsystem defines the logging code for this subsystem.
+const Subsystem = "OMSG"
+
+// log is a logger that is initialized with no output filters.  This
+// means the package will not perform any logging by default until the caller
+// requests it.
+var log btclog.Logger
+
+// The default amount of logging is none.
+func init() {
+	UseLogger(build.NewSubLogger(Subsystem, nil))
+}
+
+// DisableLog disables all library log output.  Logging output is disabled
+// by default until UseLogger is called.
+func DisableLog() {
+	UseLogger(btclog.Disabled)
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

onionmessage/onion_endpoint.go

@@ -0,0 +1,96 @@
+package onionmessage
+
+import (
+	"context"
+	"encoding/hex"
+	"log/slog"
+
+	"github.com/lightningnetwork/lnd/lnutils"
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/msgmux"
+	"github.com/lightningnetwork/lnd/subscribe"
+)
+
+// OnionMessageUpdate is onion message update dispatched to any potential
+// subscriber.
+type OnionMessageUpdate struct {
+	// Peer is the peer pubkey
+	Peer [33]byte
+
+	// PathKey is the route blinding ephemeral pubkey to be used for
+	// the onion message.
+	PathKey [33]byte
+
+	// OnionBlob is the raw serialized mix header used to relay messages in
+	// a privacy-preserving manner. This blob should be handled in the same
+	// manner as onions used to route HTLCs, with the exception that it uses
+	// blinded routes by default.
+	OnionBlob []byte
+}
+
+// OnionEndpoint handles incoming onion messages.
+type OnionEndpoint struct {
+	// subscribe.Server is used for subscriptions to onion messages.
+	onionMessageServer *subscribe.Server
+}
+
+// A compile-time check to ensure OnionEndpoint implements the Endpoint
+// interface.
+var _ msgmux.Endpoint = (*OnionEndpoint)(nil)
+
+// NewOnionEndpoint creates a new OnionEndpoint.
+func NewOnionEndpoint(messageServer *subscribe.Server) *OnionEndpoint {
+	return &OnionEndpoint{
+		onionMessageServer: messageServer,
+	}
+}
+
+// Name returns the unique name of the endpoint.
+func (o *OnionEndpoint) Name() string {
+	return "OnionMessageHandler"
+}
+
+// CanHandle checks if the endpoint can handle the incoming message.
+// It returns true if the message is an lnwire.OnionMessage.
+func (o *OnionEndpoint) CanHandle(msg msgmux.PeerMsg) bool {
+	_, ok := msg.Message.(*lnwire.OnionMessage)
+	return ok
+}
+
+// SendMessage processes the incoming onion message.
+// It returns true if the message was successfully processed.
+func (o *OnionEndpoint) SendMessage(ctx context.Context,
+	msg msgmux.PeerMsg) bool {
+
+	onionMsg, ok := msg.Message.(*lnwire.OnionMessage)
+	if !ok {
+		return false
+	}
+
+	peer := msg.PeerPub.SerializeCompressed()
+	log.DebugS(ctx, "OnionEndpoint received OnionMessage",
+		slog.String("peer", hex.EncodeToString(peer)),
+		lnutils.LogPubKey("path_key", onionMsg.PathKey),
+		lnutils.LogBytesPreview("onion_blob", onionMsg.OnionBlob),
+		slog.Int("blob length", len(onionMsg.OnionBlob)))
+
+	var peerArr [33]byte
+	copy(peerArr[:], peer)
+
+	// Convert blinding point []byte to [33]byte.
+	blinding := onionMsg.PathKey.SerializeCompressed()
+	var blindingArr [33]byte
+	copy(blindingArr[:], blinding)
+
+	err := o.onionMessageServer.SendUpdate(&OnionMessageUpdate{
+		Peer:      peerArr,
+		PathKey:   blindingArr,
+		OnionBlob: onionMsg.OnionBlob,
+	})
+	if err != nil {
+		log.ErrorS(ctx, "Failed to send onion message update", err)
+		return false
+	}
+
+	return true
+}

peer/brontide.go

@@ -47,6 +47,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/msgmux"
 	"github.com/lightningnetwork/lnd/netann"
+	"github.com/lightningnetwork/lnd/onionmessage"
 	"github.com/lightningnetwork/lnd/pool"
 	"github.com/lightningnetwork/lnd/protofsm"
 	"github.com/lightningnetwork/lnd/queue"
@@ -463,6 +464,10 @@ type Config struct {
 	// related wire messages.
 	AuxChannelNegotiator fn.Option[lnwallet.AuxChannelNegotiator]
 
+	// OnionMessageServer is an instance of a message server that dispatches
+	// onion messages to subscribers.
+	OnionMessageServer *subscribe.Server
+
 	// ShouldFwdExpEndorsement is a closure that indicates whether
 	// experimental endorsement signals should be set.
 	ShouldFwdExpEndorsement func() bool
@@ -898,6 +903,21 @@ func (p *Brontide) Start() error {
 		return fmt.Errorf("unable to load channels: %w", err)
 	}
 
+	onionMessageEndpoint := onionmessage.NewOnionEndpoint(
+		p.cfg.OnionMessageServer,
+	)
+
+	// We register the onion message endpoint with the message router.
+	err = fn.MapOptionZ(p.msgRouter, func(r msgmux.Router) error {
+		_ = r.UnregisterEndpoint(onionMessageEndpoint.Name())
+
+		return r.RegisterEndpoint(onionMessageEndpoint)
+	})
+	if err != nil {
+		return fmt.Errorf("unable to register endpoint for onion "+
+			"messaging: %w", err)
+	}
+
 	p.startTime = time.Now()
 
 	// Before launching the writeHandler goroutine, we send any channel