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.

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,80 @@
+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()
+	// Create a random payload. The content doesn't matter for this and
+	// doesn't need to be encrypted. It's also of arbitrary length, so it
+	// doesn't follow the BOLT 4 spec for onion message payload length of
+	// either 1300 or 32768 bytes. Here we just use a few bytes to keep it
+	// simple.
+	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,101 @@
+package onionmessage
+
+import (
+	"context"
+	"encoding/hex"
+	"log/slog"
+
+	"github.com/btcsuite/btclog/v2"
+	"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()
+
+	logCtx := btclog.WithCtx(ctx,
+		slog.String("peer", hex.EncodeToString(peer)),
+		lnutils.LogPubKey("path_key", onionMsg.PathKey),
+	)
+
+	log.DebugS(logCtx, "OnionEndpoint received OnionMessage",
+		btclog.HexN("onion_blob", onionMsg.OnionBlob, 10),
+		slog.Int("blob_length", len(onionMsg.OnionBlob)))
+
+	var peerArr [33]byte
+	copy(peerArr[:], peer)
+
+	// Convert path key []byte to [33]byte.
+	pathKey := onionMsg.PathKey.SerializeCompressed()
+	var pathKeyArr [33]byte
+	copy(pathKeyArr[:], pathKey)
+
+	err := o.onionMessageServer.SendUpdate(&OnionMessageUpdate{
+		Peer:      peerArr,
+		PathKey:   pathKeyArr,
+		OnionBlob: onionMsg.OnionBlob,
+	})
+	if err != nil {
+		log.ErrorS(logCtx, "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

rpcserver.go

@@ -70,6 +70,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnwallet/chanfunding"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/macaroons"
+	"github.com/lightningnetwork/lnd/onionmessage"
 	paymentsdb "github.com/lightningnetwork/lnd/payments/db"
 	"github.com/lightningnetwork/lnd/peer"
 	"github.com/lightningnetwork/lnd/peernotifier"
@@ -573,6 +574,14 @@ func MainRPCServerPermissions() map[string][]bakery.Op {
 			Entity: "offchain",
 			Action: "read",
 		}},
+		"/lnrpc.Lightning/SendOnionMessage": {{
+			Entity: "offchain",
+			Action: "write",
+		}},
+		"/lnrpc.Lightning/SubscribeOnionMessages": {{
+			Entity: "offchain",
+			Action: "read",
+		}},
 		"/lnrpc.Lightning/LookupHtlcResolution": {{
 			Entity: "offchain",
 			Action: "read",
@@ -9301,6 +9310,76 @@ func (r *rpcServer) SubscribeCustomMessages(
 	}
 }
 
+// SendOnionMessage sends a custom peer message.
+func (r *rpcServer) SendOnionMessage(ctx context.Context,
+	req *lnrpc.SendOnionMessageRequest) (*lnrpc.SendOnionMessageResponse,
+	error) {
+
+	// First we'll validate the string passed in within the request to
+	// ensure that it's a valid hex-string, and also a valid compressed
+	// public key.
+	pathKey, err := btcec.ParsePubKey(req.PathKey)
+	if err != nil {
+		return nil, fmt.Errorf("unable to decode path key bytes: %w",
+			err)
+	}
+
+	peer, err := route.NewVertexFromBytes(req.Peer)
+	if err != nil {
+		return nil, err
+	}
+
+	err = r.server.SendOnionMessage(ctx, peer, pathKey, req.Onion)
+	switch {
+	case errors.Is(err, ErrPeerNotConnected):
+		return nil, status.Error(codes.NotFound, err.Error())
+	case err != nil:
+		return nil, err
+	}
+
+	return &lnrpc.SendOnionMessageResponse{
+		Status: "onion message sent successfully",
+	}, nil
+}
+
+// SubscribeOnionMessages subscribes to a stream of incoming onion messages.
+func (r *rpcServer) SubscribeOnionMessages(
+	_ *lnrpc.SubscribeOnionMessagesRequest,
+	server lnrpc.Lightning_SubscribeOnionMessagesServer) error {
+
+	client, err := r.server.SubscribeOnionMessages()
+	if err != nil {
+		return err
+	}
+	defer client.Cancel()
+
+	for {
+		select {
+		case <-client.Quit():
+			return errors.New("shutdown")
+
+		case <-server.Context().Done():
+			return server.Context().Err()
+
+		case update := <-client.Updates():
+			oMsg, ok := update.(*onionmessage.OnionMessageUpdate)
+			if !ok {
+				return fmt.Errorf("onion message update "+
+					"failed type assertion: %T", update)
+			}
+
+			err := server.Send(&lnrpc.OnionMessage{
+				Peer:    oMsg.Peer[:],
+				PathKey: oMsg.PathKey[:],
+				Onion:   oMsg.OnionBlob,
+			})
+			if err != nil {
+				return err
+			}
+		}
+	}
+}
+
 // ListAliases returns the set of all aliases we have ever allocated along with
 // their base SCIDs and possibly a separate confirmed SCID in the case of
 // zero-conf.

server.go

@@ -422,6 +422,8 @@ type server struct {
 
 	customMessageServer *subscribe.Server
 
+	onionMessageServer *subscribe.Server
+
 	// txPublisher is a publisher with fee-bumping capability.
 	txPublisher *sweep.TxPublisher
 
@@ -727,6 +729,8 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
 
 		customMessageServer: subscribe.NewServer(),
 
+		onionMessageServer: subscribe.NewServer(),
+
 		tlsManager: tlsManager,
 
 		featureMgr: featureMgr,
@@ -2134,6 +2138,12 @@ func (s *server) Start(ctx context.Context) error {
 			return
 		}
 
+		cleanup = cleanup.add(s.onionMessageServer.Stop)
+		if err := s.onionMessageServer.Start(); err != nil {
+			startErr = err
+			return
+		}
+
 		if s.hostAnn != nil {
 			cleanup = cleanup.add(s.hostAnn.Stop)
 			if err := s.hostAnn.Start(); err != nil {
@@ -4193,6 +4203,11 @@ func (s *server) SubscribeCustomMessages() (*subscribe.Client, error) {
 	return s.customMessageServer.Subscribe()
 }
 
+// SubscribeOnionMessages subscribes to a stream of incoming onion messages.
+func (s *server) SubscribeOnionMessages() (*subscribe.Client, error) {
+	return s.onionMessageServer.Subscribe()
+}
+
 // notifyOpenChannelPeerEvent updates the access manager's maps and then calls
 // the channelNotifier's NotifyOpenChannelEvent.
 func (s *server) notifyOpenChannelPeerEvent(op wire.OutPoint,
@@ -4365,6 +4380,7 @@ func (s *server) peerConnected(conn net.Conn, connReq *connmgr.ConnReq,
 		HtlcNotifier:            s.htlcNotifier,
 		TowerClient:             towerClient,
 		DisconnectPeer:          s.DisconnectPeer,
+		OnionMessageServer:      s.onionMessageServer,
 		GenNodeAnnouncement: func(...netann.NodeAnnModifier) (
 			lnwire.NodeAnnouncement1, error) {
 
@@ -5274,6 +5290,36 @@ func (s *server) SendCustomMessage(peerPub [33]byte, msgType lnwire.MessageType,
 	return peer.SendMessageLazy(true, msg)
 }
 
+// SendOnionMessage sends a custom message to the peer with the specified
+// pubkey.
+// TODO(gijs): change this message to include path finding.
+func (s *server) SendOnionMessage(ctx context.Context, peerPub [33]byte,
+	pathKey *btcec.PublicKey, onion []byte) error {
+
+	peer, err := s.FindPeerByPubStr(string(peerPub[:]))
+	if err != nil {
+		return err
+	}
+
+	// We'll wait until the peer is active, but also listen for
+	// cancellation.
+	select {
+	case <-peer.ActiveSignal():
+	case <-peer.QuitSignal():
+		return fmt.Errorf("peer %x disconnected", peerPub)
+	case <-s.quit:
+		return ErrServerShuttingDown
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	msg := lnwire.NewOnionMessage(pathKey, onion)
+
+	// Send the message as low-priority. For now we assume that all
+	// application-defined message are low priority.
+	return peer.SendMessageLazy(true, msg)
+}
+
 // newSweepPkScriptGen creates closure that generates a new public key script
 // which should be used to sweep any funds into the on-chain wallet.
 // Specifically, the script generated is a version 0, pay-to-witness-pubkey-hash