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

@@ -511,6 +511,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()
+	msgBlindingPoint := randomPub.SerializeCompressed()
+	msgOnion := []byte{1, 2, 3}
+
+	// Send it from Bob to Alice.
+	bobMsg := &lnrpc.SendOnionMessageRequest{
+		Peer:          alice.PubKey[:],
+		BlindingPoint: msgBlindingPoint,
+		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, msgBlindingPoint, msg.BlindingPoint, "msg "+
+			"blinding point 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

@@ -666,6 +666,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) {
@@ -698,6 +700,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/onion_message"
 	"github.com/lightningnetwork/lnd/peer"
 	"github.com/lightningnetwork/lnd/peernotifier"
 	"github.com/lightningnetwork/lnd/protofsm"
@@ -207,6 +208,7 @@ func SetupLoggers(root *build.SubLoggerManager, interceptor signal.Interceptor)
 	AddSubLogger(root, chainio.Subsystem, interceptor, chainio.UseLogger)
 	AddSubLogger(root, msgmux.Subsystem, interceptor, msgmux.UseLogger)
 	AddSubLogger(root, sqldb.Subsystem, interceptor, sqldb.UseLogger)
+	AddSubLogger(root, onion_message.Subsystem, interceptor, onion_message.UseLogger)
 }
 
 // AddSubLogger is a helper method to conveniently create and register the

onion_message/log.go

@@ -0,0 +1,32 @@
+package onion_message
+
+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
+}

onion_message/onion_endpoint.go

@@ -0,0 +1,85 @@
+package onion_message
+
+import (
+	"context"
+
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/msgmux"
+	"github.com/lightningnetwork/lnd/subscribe"
+)
+
+// OnionEndpoint handles incoming onion messages.
+type OnionEndpoint struct {
+	// subscribe.Server is used for subscriptions to onion messages.
+	onionMessageServer *subscribe.Server
+}
+
+// OnionMessageUpdate is onion message update dispatched to any potential
+// subscriber.
+type OnionMessageUpdate struct {
+	// Peer is the peer pubkey
+	Peer [33]byte
+
+	// BlindingPoint is the route blinding ephemeral pubkey to be used for
+	// the onion message.
+	BlindingPoint []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
+}
+
+// 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.Debugf("OnionEndpoint received OnionMessage from peer %s: "+
+		"BlindingPoint=%v, OnionPacket[:10]=%x...", peer,
+		onionMsg.BlindingPoint, onionMsg.OnionBlob)
+
+	var peerArr [33]byte
+	copy(peerArr[:], peer)
+	err := o.onionMessageServer.SendUpdate(&OnionMessageUpdate{
+		Peer:          peerArr,
+		BlindingPoint: onionMsg.BlindingPoint.SerializeCompressed(),
+		OnionBlob:     onionMsg.OnionBlob,
+	})
+	if err != nil {
+		log.Errorf("Failed to send onion message update: %v", err)
+		return false
+	}
+
+	return true
+}
+
+// A compile-time check to ensure OnionEndpoint implements the Endpoint
+// interface.
+var _ msgmux.Endpoint = (*OnionEndpoint)(nil)

peer/brontide.go

@@ -46,6 +46,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/msgmux"
 	"github.com/lightningnetwork/lnd/netann"
+	"github.com/lightningnetwork/lnd/onion_message"
 	"github.com/lightningnetwork/lnd/pool"
 	"github.com/lightningnetwork/lnd/protofsm"
 	"github.com/lightningnetwork/lnd/queue"
@@ -451,6 +452,10 @@ type Config struct {
 	// used to modify the way the co-op close transaction is constructed.
 	AuxChanCloser fn.Option[chancloser.AuxChanCloser]
 
+	// 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
@@ -886,6 +891,21 @@ func (p *Brontide) Start() error {
 		return fmt.Errorf("unable to load channels: %w", err)
 	}
 
+	onionMessageEndpoint := onion_message.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