Bolt8 transport

electrumx/server/env.py

@@ -30,7 +30,7 @@ class Env(EnvBase):
     # Peer discovery
     PD_OFF, PD_SELF, PD_ON = ('OFF', 'SELF', 'ON')
     SSL_PROTOCOLS = {'ssl', 'wss'}
-    KNOWN_PROTOCOLS = {'ssl', 'tcp', 'ws', 'wss', 'rpc'}
+    KNOWN_PROTOCOLS = {'ssl', 'tcp', 'ws', 'wss', 'rpc', 'bolt8'}
 
     coin: Type[Coin]
 
@@ -43,6 +43,9 @@ def __init__(self, coin=None):
 
         # Core items
 
+        self.is_public = self.boolean('PUBLIC', False)
+        self.is_watchtower = self.boolean('WATCHTOWER', False)
+        assert not (self.is_public and self.is_watchtower)
         self.db_dir = self.required('DB_DIRECTORY')
         self.daemon_url = self.required('DAEMON_URL')
         if coin is not None:
@@ -54,7 +57,6 @@ def __init__(self, coin=None):
             self.coin = Coin.lookup_coin_class(coin_name, network)
 
         # Peer discovery
-
         self.peer_discovery = self.peer_discovery_enum()
         self.peer_announce = self.boolean('PEER_ANNOUNCE', True)
         self.force_proxy = self.boolean('FORCE_PROXY', False)
@@ -99,6 +101,8 @@ def __init__(self, coin=None):
         if {service.protocol for service in self.services}.intersection(self.SSL_PROTOCOLS):
             self.ssl_certfile = self.required('SSL_CERTFILE')
             self.ssl_keyfile = self.required('SSL_KEYFILE')
+        if any([service.protocol == 'bolt8' for service in self.services]):
+            self.bolt8_keyfile = self.required('BOLT8_KEYFILE')
         self.report_services = self.services_to_report()
 
     def sane_max_sessions(self):

electrumx/server/session.py

@@ -164,9 +164,10 @@ def __init__(
         # Event triggered when electrumx is listening for incoming requests.
         self.server_listening = Event()
         self.session_event = Event()
+        self._authorized_users = self._read_users()
 
         # Set up the RPC request handlers
-        cmds = ('add_peer daemon_url disconnect getinfo groups log peers '
+        cmds = ('add_peer add_user rm_user daemon_url disconnect getinfo groups log peers '
                 'query reorg sessions stop debug_memusage_list_all_objects '
                 'debug_memusage_get_random_backref_chain'.split())
         LocalRPC.request_handlers = {cmd: getattr(self, 'rpc_' + cmd)
@@ -178,6 +179,57 @@ def _ssl_context(self):
             self._sslc.load_cert_chain(self.env.ssl_certfile, keyfile=self.env.ssl_keyfile)
         return self._sslc
 
+    def _get_bolt8_server(self):
+        from electrum import ecc
+        from electrum import logging
+        from electrum.lntransport import create_bolt8_server
+        logging._configure_stderr_logging(verbosity='*')
+        path = os.path.join(self.env.bolt8_keyfile)
+        if os.path.exists(path):
+            with open(path, 'r') as f:
+                s = f.read()
+            self.bolt8_privkey = bytes.fromhex(s)
+        else:
+            self.bolt8_privkey = os.urandom(32)
+            with open(path, 'w') as f:
+                f.write(self.bolt8_privkey.hex())
+        self.bolt8_pubkey = ecc.ECPrivkey(self.bolt8_privkey).get_public_key_bytes()
+        self.logger.info(f'public server: {self.env.is_public}')
+        self.logger.info(f'bolt8 pubkey {self.bolt8_pubkey.hex()}')
+        whitelist = None if self.env.is_public else self._authorized_users
+        # allow self, for watchtower
+        if whitelist is not None:
+            whitelist.add(self.bolt8_pubkey)
+        return partial(create_bolt8_server, b'electrum', self.bolt8_privkey, whitelist)
+
+    def add_user(self, pubkey):
+        assert len(pubkey) == 33
+        self._authorized_users.add(pubkey)
+        self._save_users()
+
+    def rm_user(self, pubkey):
+        assert len(pubkey) == 33
+        self._authorized_users.remove(pubkey)
+        self._save_users()
+
+    def _save_users(self):
+        import json
+        path = os.path.join(self.env.db_dir, 'authorized_users')
+        s = json.dumps([x.hex() for x in sorted(self._authorized_users)])
+        with open(path, 'w') as f:
+            f.write(s)
+
+    def _read_users(self):
+        import json
+        path = os.path.join(self.env.db_dir, 'authorized_users')
+        if os.path.exists(path):
+            with open(path, 'r') as f:
+                _list = json.loads(f.read())
+                _set = set([bytes.fromhex(x) for x in _list])
+        else:
+            _set = set()
+        return _set
+
     async def _start_servers(self, services):
         for service in services:
             kind = service.protocol.upper()
@@ -191,6 +243,8 @@ async def _start_servers(self, services):
                 session_class = self.env.coin.SESSIONCLS
             if service.protocol in ('ws', 'wss'):
                 serve = serve_ws
+            elif service.protocol in ('bolt8'):
+                serve = self._get_bolt8_server()
             else:
                 serve = serve_rs
             # FIXME: pass the service not the kind
@@ -434,6 +488,18 @@ async def rpc_add_peer(self, real_name):
         await self.peer_mgr.add_localRPC_peer(real_name)
         return f"peer '{real_name}' added"
 
+    async def rpc_add_user(self, pubkey):
+        '''Add a whitelisted user.
+        '''
+        self.add_user(bytes.fromhex(pubkey))
+        return f"user added"
+
+    async def rpc_rm_user(self, pubkey):
+        '''Remove a whitelisted user.
+        '''
+        self.rm_user(bytes.fromhex(pubkey))
+        return f"user removed"
+
     async def rpc_disconnect(self, session_ids):
         '''Disconnect sesssions.
 
@@ -627,6 +693,41 @@ async def rpc_debug_memusage_get_random_backref_chain(self, objtype: str) -> str
                     output=fd))
             return fd.getvalue()
 
+    async def start_watchtower(self):
+        # FIXME: this creates a socket to self.
+        # We should create a Network object that taps directly into ElectrumX RPC
+        import electrum
+        for s in self.env.services:
+            if s.protocol == 'bolt8':
+                server_addr = 'localhost:%d:b:%s'%(s.port, self.bolt8_pubkey.hex())
+                break
+        else:
+            return
+        electrum.logging._configure_stderr_logging(verbosity='*')
+        electrum.util._asyncio_event_loop = asyncio.get_event_loop()
+        config = {
+            'server': server_addr,
+            'oneserver': True,
+        }
+        if self.env.coin.NET == 'regtest':
+            electrum.constants.set_regtest()
+            config['regtest'] = True
+        elif self.env.coin.NET == 'testnet':
+            electrum.constants.set_regtest()
+            config['testnet'] = True
+        config = electrum.simple_config.SimpleConfig(config)
+        config.set_bolt8_privkey_for_server(self.bolt8_pubkey.hex(), self.bolt8_privkey.hex())
+
+        self.network = electrum.network.Network(config)
+        self.network.start()
+        self.lnwatcher = electrum.lnwatcher.WatchTower(self.network)
+        self.lnwatcher.adb.start_network(self.network)
+        await self.lnwatcher.start_watching()
+
+    async def stop_watchtower(self):
+        await self.lnwatcher.stop()
+        await self.network.stop()
+
     # --- External Interface
 
     async def serve(self, notifications, event):
@@ -665,6 +766,10 @@ async def serve(self, notifications, event):
             # Start notifications; initialize hsub_results
             await notifications.start(self.db.db_height, self._notify_sessions)
             await self._start_external_servers()
+            # start watchtower
+            if self.env.is_watchtower:
+                self.logger.info('starting watchtower')
+                await self.start_watchtower()
             # Peer discovery should start after the external servers
             # because we connect to ourself
             async with self._task_group as group:
@@ -1429,6 +1534,12 @@ async def server_version(self, client_name='', protocol_version=None):
 
         return electrumx.version, self.protocol_version_string()
 
+    async def watchtower_get_ctn(self, outpoint, addr):
+        return await self.session_mgr.lnwatcher.get_ctn(outpoint, addr)
+
+    async def watchtower_add_sweep_tx(self, *args):
+        return await self.session_mgr.lnwatcher.sweepstore.add_sweep_tx(*args)
+
     async def crash_old_client(self, ptuple, crash_client_ver):
         if crash_client_ver:
             client_ver = util.protocol_tuple(self.client)
@@ -1551,6 +1662,8 @@ def set_request_handlers(self, ptuple):
             'server.peers.subscribe': self.peers_subscribe,
             'server.ping': self.ping,
             'server.version': self.server_version,
+            'watchtower.get_ctn': self.watchtower_get_ctn,
+            'watchtower.add_sweep_tx': self.watchtower_add_sweep_tx,
         }
 
         if ptuple >= (1, 4, 2):

electrumx_rpc

@@ -42,6 +42,22 @@ other_commands = {
             'help': 'e.g. "a.domain.name s995 t"',
         },
     ),
+    'add_user': (
+        'add a private user',
+        [], {
+            'type': str,
+            'dest': 'pubkey',
+            'help': 'public key"',
+        },
+    ),
+    'rm_user': (
+        'remove a private user',
+        [], {
+            'type': str,
+            'dest': 'pubkey',
+            'help': 'public key"',
+        },
+    ),
     'daemon_url': (
         "replace the daemon's URL at run-time, and forecefully rotate "
         " to the first URL in the list",