lnworker, lnpeer: add OnionMessageManager to lnworker and call from l…

…npeer on onion message arrival

electrum/lnpeer.py

@@ -30,8 +30,8 @@
 from .transaction import PartialTxOutput, match_script_against_template, Sighash
 from .logging import Logger
 from .lnrouter import RouteEdge
-from .lnonion import (new_onion_packet, OnionFailureCode, calc_hops_data_for_payment,
-                      process_onion_packet, OnionPacket, construct_onion_error, obfuscate_onion_error, OnionRoutingFailure,
+from .lnonion import (new_onion_packet, OnionFailureCode, calc_hops_data_for_payment, process_onion_packet,
+                      OnionPacket, construct_onion_error, obfuscate_onion_error, OnionRoutingFailure,
                       ProcessedOnionPacket, UnsupportedOnionPacketVersion, InvalidOnionMac, InvalidOnionPubkey,
                       OnionFailureCodeMetaFlag)
 from .lnchannel import Channel, RevokeAndAck, RemoteCtnTooFarInFuture, ChannelState, PeerState, ChanCloseOption, CF_ANNOUNCE_CHANNEL
@@ -2863,3 +2863,7 @@ def process_onion_packet(
         if self.network.config.TEST_FAIL_HTLCS_WITH_TEMP_NODE_FAILURE:
             raise OnionRoutingFailure(code=OnionFailureCode.TEMPORARY_NODE_FAILURE, data=b'')
         return processed_onion
+
+    def on_onion_message(self, payload):
+        if hasattr(self.lnworker, 'onion_message_manager'):  # only on LNWallet
+            self.lnworker.onion_message_manager.on_onion_message(payload)

electrum/lnworker.py

@@ -31,6 +31,7 @@
 
 from . import constants, util
 from . import keystore
+from .onion_message import OnionMessageManager
 from .util import profiler, chunks, OldTaskGroup, ESocksProxy
 from .invoices import Invoice, PR_UNPAID, PR_EXPIRED, PR_PAID, PR_INFLIGHT, PR_FAILED, PR_ROUTING, LN_EXPIRY_NEVER
 from .invoices import BaseInvoice
@@ -882,6 +883,7 @@ def __init__(self, wallet: 'Abstract_Wallet', xprv):
 
         self.nostr_keypair = generate_keypair(BIP32Node.from_xkey(xprv), LnKeyFamily.NOSTR_KEY)
         self.swap_manager = SwapManager(wallet=self.wallet, lnworker=self)
+        self.onion_message_manager = OnionMessageManager(self)
 
     def has_deterministic_node_id(self) -> bool:
         return bool(self.db.get('lightning_xprv'))
@@ -970,6 +972,7 @@ def start_network(self, network: 'Network'):
         self.lnwatcher = LNWalletWatcher(self, network)
         self.swap_manager.start_network(network)
         self.lnrater = LNRater(self, network)
+        self.onion_message_manager.start_network(network=network)
 
         for chan in self.channels.values():
             if chan.need_to_subscribe():
@@ -1000,6 +1003,8 @@ async def stop(self):
             self.lnwatcher = None
         if self.swap_manager and self.swap_manager.network:  # may not be present in tests
             await self.swap_manager.stop()
+        if self.onion_message_manager:
+            await self.onion_message_manager.stop()
 
     async def wait_for_received_pending_htlcs_to_get_removed(self):
         assert self.stopping_soon is True

tests/test_lnpeer.py

@@ -156,6 +156,7 @@ def __init__(self, *, local_keypair: Keypair, chans: Iterable['Channel'], tx_que
         self.taskgroup = OldTaskGroup()
         self.lnwatcher = None
         self.swap_manager = None
+        self.onion_message_manager = None
         self.listen_server = None
         self._channels = {chan.channel_id: chan for chan in chans}
         self.payment_info = {}