lnworker, lnpeer: add OnionMessageManager to LNWallet, add on_onion_m…

…essage handler

to lnpeer and dispatch to OnionMessageManager

electrum/lnpeer.py

@@ -17,7 +17,7 @@
 from aiorpcx import ignore_after
 
 from .crypto import sha256, sha256d
-from . import bitcoin, util
+from . import bitcoin, util, onion_message
 from . import ecc
 from .ecc import ecdsa_sig64_from_r_and_s, ecdsa_der_sig_from_ecdsa_sig64, ECPubkey
 from . import constants
@@ -28,8 +28,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
@@ -2781,3 +2781,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 getattr(self.lnworker, 'onion_message_manager'):  # only on LNWallet
+            self.lnworker.onion_message_manager.on_onion_message(payload)

electrum/lnworker.py

@@ -30,6 +30,7 @@
 
 from . import constants, util
 from . import keystore
+from .onion_message import create_reply_path, OnionMessageManager
 from .util import profiler, chunks, OldTaskGroup
 from .invoices import Invoice, PR_UNPAID, PR_EXPIRED, PR_PAID, PR_INFLIGHT, PR_FAILED, PR_ROUTING, LN_EXPIRY_NEVER
 from .invoices import BaseInvoice
@@ -858,6 +859,7 @@ def __init__(self, wallet: 'Abstract_Wallet', xprv):
         self.payment_bundles = []                       # lists of hashes. todo:persist
         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'))
@@ -946,6 +948,7 @@ def start_network(self, network: 'Network'):
         self.lnwatcher = LNWalletWatcher(self, network)
         self.swap_manager.start_network(network=network, lnwatcher=self.lnwatcher)
         self.lnrater = LNRater(self, network)
+        self.onion_message_manager.start_network(network=network)
 
         for chan in self.channels.values():
             if chan.need_to_subscribe():
@@ -976,6 +979,8 @@ async def stop(self):
             self.lnwatcher = None
         if self.swap_manager:  # 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