diff --git a/itest/list_on_test.go b/itest/list_on_test.go index 92c6547b3..3fc0fba43 100644 --- a/itest/list_on_test.go +++ b/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, diff --git a/itest/lnd_onion_message_test.go b/itest/lnd_onion_message_test.go new file mode 100644 index 000000000..e0f9c7471 --- /dev/null +++ b/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) + } +} diff --git a/lntest/rpc/lnd.go b/lntest/rpc/lnd.go index 265aab91a..946b37b88 100644 --- a/lntest/rpc/lnd.go +++ b/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( diff --git a/log.go b/log.go index 2484a35dd..5f80bb7f5 100644 --- a/log.go +++ b/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 diff --git a/onionmessage/log.go b/onionmessage/log.go new file mode 100644 index 000000000..7bf093681 --- /dev/null +++ b/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 +} diff --git a/onionmessage/onion_endpoint.go b/onionmessage/onion_endpoint.go new file mode 100644 index 000000000..ef1c5ecc0 --- /dev/null +++ b/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 +} diff --git a/peer/brontide.go b/peer/brontide.go index 9191cbb2e..4a196fb95 100644 --- a/peer/brontide.go +++ b/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 diff --git a/rpcserver.go b/rpcserver.go index d3d3c5180..fd6743407 100644 --- a/rpcserver.go +++ b/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. diff --git a/server.go b/server.go index 6d0b28f7e..b7d4f2370 100644 --- a/server.go +++ b/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