remote watchtower: initial commit

electrum/daemon.py

@@ -33,7 +33,7 @@ from typing import Dict, Optional, Tuple
 
 import jsonrpclib
 
-from .jsonrpc import VerifyingJSONRPCServer
+from .jsonrpc import SimpleJSONRPCServer, PasswordProtectedJSONRPCServer
 from .version import ELECTRUM_VERSION
 from .network import Network
 from .util import (json_decode, DaemonThread, to_string,
@@ -147,6 +147,8 @@ class Daemon(DaemonThread):
         self.server = None
         if listen_jsonrpc:
             self.init_server(config, fd)
+        if config.get('watchtower_host'):
+            self.init_watchtower()
         self.start()
 
     def init_server(self, config: SimpleConfig, fd):
@@ -154,8 +156,9 @@ class Daemon(DaemonThread):
         port = config.get('rpcport', 0)
         rpc_user, rpc_password = get_rpc_credentials(config)
         try:
-            server = VerifyingJSONRPCServer((host, port), logRequests=False,
-                                            rpc_user=rpc_user, rpc_password=rpc_password)
+            server = PasswordProtectedJSONRPCServer(
+                (host, port), logRequests=False,
+                rpc_user=rpc_user, rpc_password=rpc_password)
         except Exception as e:
             self.logger.error(f'cannot initialize RPC server on host {host}: {repr(e)}')
             self.server = None
@@ -173,6 +176,12 @@ class Daemon(DaemonThread):
             server.register_function(getattr(self.cmd_runner, cmdname), cmdname)
         server.register_function(self.run_cmdline, 'run_cmdline')
 
+    def init_watchtower(self):
+        host = self.config.get('watchtower_host')
+        port = self.config.get('watchtower_port', 12345)
+        server = SimpleJSONRPCServer((host, port), logRequests=False)
+        server.register_function(self.network.lnwatcher, 'add_sweep_tx')
+
     def ping(self):
         return True
 

electrum/jsonrpc.py

@@ -1,3 +1,4 @@
+
 #!/usr/bin/env python3
 #
 # Electrum - lightweight Bitcoin client
@@ -48,12 +49,10 @@ class RPCAuthUnsupportedType(Exception):
 
 
 # based on http://acooke.org/cute/BasicHTTPA0.html by andrew cooke
-class VerifyingJSONRPCServer(SimpleJSONRPCServer, Logger):
+class AuthenticatedJSONRPCServer(SimpleJSONRPCServer, Logger):
 
-    def __init__(self, *args, rpc_user, rpc_password, **kargs):
+    def __init__(self, *args, **kargs):
         Logger.__init__(self)
-        self.rpc_user = rpc_user
-        self.rpc_password = rpc_password
 
         class VerifyingRequestHandler(SimpleJSONRPCRequestHandler):
             def parse_request(myself):
@@ -73,10 +72,20 @@ class VerifyingJSONRPCServer(SimpleJSONRPCServer, Logger):
                         self.logger.exception('')
                         myself.send_error(500, repr(e))
                 return False
-
         SimpleJSONRPCServer.__init__(
             self, requestHandler=VerifyingRequestHandler, *args, **kargs)
 
+    def authenticate(self, headers):
+        raise Exception('undefined')
+
+
+class PasswordProtectedJSONRPCServer(AuthenticatedJSONRPCServer):
+
+    def __init__(self, *args, rpc_user, rpc_password, **kargs):
+        self.rpc_user = rpc_user
+        self.rpc_password = rpc_password
+        AuthenticatedJSONRPCServer.__init__(self, *args, **kargs)
+
     def authenticate(self, headers):
         if self.rpc_password == '':
             # RPC authentication is disabled
@@ -97,3 +106,21 @@ class VerifyingJSONRPCServer(SimpleJSONRPCServer, Logger):
                 and util.constant_time_compare(password, self.rpc_password)):
             time.sleep(0.050)
             raise RPCAuthCredentialsInvalid()
+
+
+class AccountsJSONRPCServer(AuthenticatedJSONRPCServer):
+    """ user accounts """
+
+    def __init__(self, *args, **kargs):
+        self.users = {}
+        AuthenticatedJSONRPCServer.__init__(self, *args, **kargs)
+        self.register_function(self.add_user, 'add_user')
+
+    def authenticate(self, headers):
+        # todo: verify signature
+        return
+
+    def add_user(self, pubkey):
+        user_id = len(self.users)
+        self.users[user_id] = pubkey
+        return user_id

electrum/lnwatcher.py

@@ -2,6 +2,8 @@ import threading
 from typing import NamedTuple, Iterable
 import os
 from collections import defaultdict
+import asyncio
+import jsonrpclib
 
 from .util import PrintError, bh2u, bfh, NoDynamicFeeEstimates, aiosafe
 from .lnutil import EncumberedTransaction, Outpoint
@@ -20,7 +22,7 @@ class LNWatcher(PrintError):
 
     def __init__(self, network):
         self.network = network
-
+        self.config = network.config
         path = os.path.join(network.config.path, "watcher_db")
         storage = WalletStorage(path)
         self.addr_sync = AddressSynchronizer(storage)
@@ -41,6 +43,25 @@ class LNWatcher(PrintError):
 
         self.network.register_callback(self.on_network_update,
                                        ['network_updated', 'blockchain_updated', 'verified', 'wallet_updated'])
+        # remote watchtower
+        watchtower_url = self.config.get('watchtower_url')
+        self.watchtower = jsonrpclib.Server(watchtower_url) if watchtower_url else None
+        self.watchtower_queue = asyncio.Queue()
+        asyncio.run_coroutine_threadsafe(self.watchtower_task(), self.network.asyncio_loop)
+
+    def with_watchtower(func):
+        def wrapper(self, *args, **kwargs):
+            if self.watchtower:
+                self.watchtower_queue.put_nowait((func.__name__, args, kwargs))
+            return func(self, *args, **kwargs)
+        return wrapper
+
+    async def watchtower_task(self):
+        while True:
+            name, args, kwargs = await self.watchtower_queue.get()
+            self.print_error('sending to watchtower', name, args)
+            func = getattr(self.watchtower, name)
+            func(*args, **kwargs)
 
     def write_to_disk(self):
         # FIXME: json => every update takes linear instead of constant disk write
@@ -151,6 +172,7 @@ class LNWatcher(PrintError):
                                      .format(num_conf, e_tx.csv_delay, funding_outpoint, ctx.txid()))
         return keep_watching_this
 
+    @with_watchtower
     def add_sweep_tx(self, funding_outpoint: str, ctx_txid: str, encumbered_sweeptx: EncumberedTransaction):
         if encumbered_sweeptx is None:
             return