lnwatcher.py (18166B)
1 # Copyright (C) 2018 The Electrum developers 2 # Distributed under the MIT software license, see the accompanying 3 # file LICENCE or http://www.opensource.org/licenses/mit-license.php 4 5 from typing import NamedTuple, Iterable, TYPE_CHECKING 6 import os 7 import asyncio 8 from enum import IntEnum, auto 9 from typing import NamedTuple, Dict 10 11 from . import util 12 from .sql_db import SqlDB, sql 13 from .wallet_db import WalletDB 14 from .util import bh2u, bfh, log_exceptions, ignore_exceptions, TxMinedInfo, random_shuffled_copy 15 from .address_synchronizer import AddressSynchronizer, TX_HEIGHT_LOCAL, TX_HEIGHT_UNCONF_PARENT, TX_HEIGHT_UNCONFIRMED 16 from .transaction import Transaction, TxOutpoint 17 18 if TYPE_CHECKING: 19 from .network import Network 20 from .lnsweep import SweepInfo 21 from .lnworker import LNWallet 22 23 class ListenerItem(NamedTuple): 24 # this is triggered when the lnwatcher is all done with the outpoint used as index in LNWatcher.tx_progress 25 all_done : asyncio.Event 26 # txs we broadcast are put on this queue so that the test can wait for them to get mined 27 tx_queue : asyncio.Queue 28 29 class TxMinedDepth(IntEnum): 30 """ IntEnum because we call min() in get_deepest_tx_mined_depth_for_txids """ 31 DEEP = auto() 32 SHALLOW = auto() 33 MEMPOOL = auto() 34 FREE = auto() 35 36 37 create_sweep_txs=""" 38 CREATE TABLE IF NOT EXISTS sweep_txs ( 39 funding_outpoint VARCHAR(34) NOT NULL, 40 ctn INTEGER NOT NULL, 41 prevout VARCHAR(34), 42 tx VARCHAR 43 )""" 44 45 create_channel_info=""" 46 CREATE TABLE IF NOT EXISTS channel_info ( 47 outpoint VARCHAR(34) NOT NULL, 48 address VARCHAR(32), 49 PRIMARY KEY(outpoint) 50 )""" 51 52 53 class SweepStore(SqlDB): 54 55 def __init__(self, path, network): 56 super().__init__(network.asyncio_loop, path) 57 58 def create_database(self): 59 c = self.conn.cursor() 60 c.execute(create_channel_info) 61 c.execute(create_sweep_txs) 62 self.conn.commit() 63 64 @sql 65 def get_sweep_tx(self, funding_outpoint, prevout): 66 c = self.conn.cursor() 67 c.execute("SELECT tx FROM sweep_txs WHERE funding_outpoint=? AND prevout=?", (funding_outpoint, prevout)) 68 return [Transaction(bh2u(r[0])) for r in c.fetchall()] 69 70 @sql 71 def list_sweep_tx(self): 72 c = self.conn.cursor() 73 c.execute("SELECT funding_outpoint FROM sweep_txs") 74 return set([r[0] for r in c.fetchall()]) 75 76 @sql 77 def add_sweep_tx(self, funding_outpoint, ctn, prevout, raw_tx): 78 c = self.conn.cursor() 79 assert Transaction(raw_tx).is_complete() 80 c.execute("""INSERT INTO sweep_txs (funding_outpoint, ctn, prevout, tx) VALUES (?,?,?,?)""", (funding_outpoint, ctn, prevout, bfh(raw_tx))) 81 self.conn.commit() 82 83 @sql 84 def get_num_tx(self, funding_outpoint): 85 c = self.conn.cursor() 86 c.execute("SELECT count(*) FROM sweep_txs WHERE funding_outpoint=?", (funding_outpoint,)) 87 return int(c.fetchone()[0]) 88 89 @sql 90 def get_ctn(self, outpoint, addr): 91 if not self._has_channel(outpoint): 92 self._add_channel(outpoint, addr) 93 c = self.conn.cursor() 94 c.execute("SELECT max(ctn) FROM sweep_txs WHERE funding_outpoint=?", (outpoint,)) 95 return int(c.fetchone()[0] or 0) 96 97 @sql 98 def remove_sweep_tx(self, funding_outpoint): 99 c = self.conn.cursor() 100 c.execute("DELETE FROM sweep_txs WHERE funding_outpoint=?", (funding_outpoint,)) 101 self.conn.commit() 102 103 def _add_channel(self, outpoint, address): 104 c = self.conn.cursor() 105 c.execute("INSERT INTO channel_info (address, outpoint) VALUES (?,?)", (address, outpoint)) 106 self.conn.commit() 107 108 @sql 109 def remove_channel(self, outpoint): 110 c = self.conn.cursor() 111 c.execute("DELETE FROM channel_info WHERE outpoint=?", (outpoint,)) 112 self.conn.commit() 113 114 def _has_channel(self, outpoint): 115 c = self.conn.cursor() 116 c.execute("SELECT * FROM channel_info WHERE outpoint=?", (outpoint,)) 117 r = c.fetchone() 118 return r is not None 119 120 @sql 121 def get_address(self, outpoint): 122 c = self.conn.cursor() 123 c.execute("SELECT address FROM channel_info WHERE outpoint=?", (outpoint,)) 124 r = c.fetchone() 125 return r[0] if r else None 126 127 @sql 128 def list_channels(self): 129 c = self.conn.cursor() 130 c.execute("SELECT outpoint, address FROM channel_info") 131 return [(r[0], r[1]) for r in c.fetchall()] 132 133 134 135 class LNWatcher(AddressSynchronizer): 136 LOGGING_SHORTCUT = 'W' 137 138 def __init__(self, network: 'Network'): 139 AddressSynchronizer.__init__(self, WalletDB({}, manual_upgrades=False)) 140 self.config = network.config 141 self.callbacks = {} # address -> lambda: coroutine 142 self.network = network 143 util.register_callback( 144 self.on_network_update, 145 ['network_updated', 'blockchain_updated', 'verified', 'wallet_updated', 'fee']) 146 147 # status gets populated when we run 148 self.channel_status = {} 149 150 async def stop(self): 151 await super().stop() 152 util.unregister_callback(self.on_network_update) 153 154 def get_channel_status(self, outpoint): 155 return self.channel_status.get(outpoint, 'unknown') 156 157 def add_channel(self, outpoint: str, address: str) -> None: 158 assert isinstance(outpoint, str) 159 assert isinstance(address, str) 160 cb = lambda: self.check_onchain_situation(address, outpoint) 161 self.add_callback(address, cb) 162 163 async def unwatch_channel(self, address, funding_outpoint): 164 self.logger.info(f'unwatching {funding_outpoint}') 165 self.remove_callback(address) 166 167 def remove_callback(self, address): 168 self.callbacks.pop(address, None) 169 170 def add_callback(self, address, callback): 171 self.add_address(address) 172 self.callbacks[address] = callback 173 174 @log_exceptions 175 async def on_network_update(self, event, *args): 176 if event in ('verified', 'wallet_updated'): 177 if args[0] != self: 178 return 179 if not self.synchronizer: 180 self.logger.info("synchronizer not set yet") 181 return 182 for address, callback in list(self.callbacks.items()): 183 await callback() 184 185 async def check_onchain_situation(self, address, funding_outpoint): 186 # early return if address has not been added yet 187 if not self.is_mine(address): 188 return 189 spenders = self.inspect_tx_candidate(funding_outpoint, 0) 190 # inspect_tx_candidate might have added new addresses, in which case we return ealy 191 if not self.is_up_to_date(): 192 return 193 funding_txid = funding_outpoint.split(':')[0] 194 funding_height = self.get_tx_height(funding_txid) 195 closing_txid = spenders.get(funding_outpoint) 196 closing_height = self.get_tx_height(closing_txid) 197 if closing_txid: 198 closing_tx = self.db.get_transaction(closing_txid) 199 if closing_tx: 200 keep_watching = await self.do_breach_remedy(funding_outpoint, closing_tx, spenders) 201 else: 202 self.logger.info(f"channel {funding_outpoint} closed by {closing_txid}. still waiting for tx itself...") 203 keep_watching = True 204 else: 205 keep_watching = True 206 await self.update_channel_state( 207 funding_outpoint=funding_outpoint, 208 funding_txid=funding_txid, 209 funding_height=funding_height, 210 closing_txid=closing_txid, 211 closing_height=closing_height, 212 keep_watching=keep_watching) 213 if not keep_watching: 214 await self.unwatch_channel(address, funding_outpoint) 215 216 async def do_breach_remedy(self, funding_outpoint, closing_tx, spenders) -> bool: 217 raise NotImplementedError() # implemented by subclasses 218 219 async def update_channel_state(self, *, funding_outpoint: str, funding_txid: str, 220 funding_height: TxMinedInfo, closing_txid: str, 221 closing_height: TxMinedInfo, keep_watching: bool) -> None: 222 raise NotImplementedError() # implemented by subclasses 223 224 def inspect_tx_candidate(self, outpoint, n): 225 prev_txid, index = outpoint.split(':') 226 txid = self.db.get_spent_outpoint(prev_txid, int(index)) 227 result = {outpoint:txid} 228 if txid is None: 229 self.channel_status[outpoint] = 'open' 230 return result 231 if n == 0 and not self.is_deeply_mined(txid): 232 self.channel_status[outpoint] = 'closed (%d)' % self.get_tx_height(txid).conf 233 else: 234 self.channel_status[outpoint] = 'closed (deep)' 235 tx = self.db.get_transaction(txid) 236 for i, o in enumerate(tx.outputs()): 237 if not self.is_mine(o.address): 238 self.add_address(o.address) 239 elif n < 2: 240 r = self.inspect_tx_candidate(txid+':%d'%i, n+1) 241 result.update(r) 242 return result 243 244 def get_tx_mined_depth(self, txid: str): 245 if not txid: 246 return TxMinedDepth.FREE 247 tx_mined_depth = self.get_tx_height(txid) 248 height, conf = tx_mined_depth.height, tx_mined_depth.conf 249 if conf > 100: 250 return TxMinedDepth.DEEP 251 elif conf > 0: 252 return TxMinedDepth.SHALLOW 253 elif height in (TX_HEIGHT_UNCONFIRMED, TX_HEIGHT_UNCONF_PARENT): 254 return TxMinedDepth.MEMPOOL 255 elif height == TX_HEIGHT_LOCAL: 256 return TxMinedDepth.FREE 257 elif height > 0 and conf == 0: 258 # unverified but claimed to be mined 259 return TxMinedDepth.MEMPOOL 260 else: 261 raise NotImplementedError() 262 263 def is_deeply_mined(self, txid): 264 return self.get_tx_mined_depth(txid) == TxMinedDepth.DEEP 265 266 267 class WatchTower(LNWatcher): 268 269 LOGGING_SHORTCUT = 'W' 270 271 def __init__(self, network): 272 LNWatcher.__init__(self, network) 273 self.network = network 274 self.sweepstore = SweepStore(os.path.join(self.network.config.path, "watchtower_db"), network) 275 # this maps funding_outpoints to ListenerItems, which have an event for when the watcher is done, 276 # and a queue for seeing which txs are being published 277 self.tx_progress = {} # type: Dict[str, ListenerItem] 278 279 def diagnostic_name(self): 280 return "local_tower" 281 282 async def start_watching(self): 283 # I need to watch the addresses from sweepstore 284 lst = await self.sweepstore.list_channels() 285 for outpoint, address in random_shuffled_copy(lst): 286 self.add_channel(outpoint, address) 287 288 async def do_breach_remedy(self, funding_outpoint, closing_tx, spenders): 289 keep_watching = False 290 for prevout, spender in spenders.items(): 291 if spender is not None: 292 keep_watching |= not self.is_deeply_mined(spender) 293 continue 294 sweep_txns = await self.sweepstore.get_sweep_tx(funding_outpoint, prevout) 295 for tx in sweep_txns: 296 await self.broadcast_or_log(funding_outpoint, tx) 297 keep_watching = True 298 return keep_watching 299 300 async def broadcast_or_log(self, funding_outpoint: str, tx: Transaction): 301 height = self.get_tx_height(tx.txid()).height 302 if height != TX_HEIGHT_LOCAL: 303 return 304 try: 305 txid = await self.network.broadcast_transaction(tx) 306 except Exception as e: 307 self.logger.info(f'broadcast failure: txid={tx.txid()}, funding_outpoint={funding_outpoint}: {repr(e)}') 308 else: 309 self.logger.info(f'broadcast success: txid={tx.txid()}, funding_outpoint={funding_outpoint}') 310 if funding_outpoint in self.tx_progress: 311 await self.tx_progress[funding_outpoint].tx_queue.put(tx) 312 return txid 313 314 def get_ctn(self, outpoint, addr): 315 async def f(): 316 return await self.sweepstore.get_ctn(outpoint, addr) 317 return self.network.run_from_another_thread(f()) 318 319 def get_num_tx(self, outpoint): 320 async def f(): 321 return await self.sweepstore.get_num_tx(outpoint) 322 return self.network.run_from_another_thread(f()) 323 324 def list_sweep_tx(self): 325 async def f(): 326 return await self.sweepstore.list_sweep_tx() 327 return self.network.run_from_another_thread(f()) 328 329 def list_channels(self): 330 async def f(): 331 return await self.sweepstore.list_channels() 332 return self.network.run_from_another_thread(f()) 333 334 async def unwatch_channel(self, address, funding_outpoint): 335 await super().unwatch_channel(address, funding_outpoint) 336 await self.sweepstore.remove_sweep_tx(funding_outpoint) 337 await self.sweepstore.remove_channel(funding_outpoint) 338 if funding_outpoint in self.tx_progress: 339 self.tx_progress[funding_outpoint].all_done.set() 340 341 async def update_channel_state(self, *args, **kwargs): 342 pass 343 344 345 346 347 class LNWalletWatcher(LNWatcher): 348 349 def __init__(self, lnworker: 'LNWallet', network: 'Network'): 350 self.network = network 351 self.lnworker = lnworker 352 LNWatcher.__init__(self, network) 353 354 def diagnostic_name(self): 355 return f"{self.lnworker.wallet.diagnostic_name()}-LNW" 356 357 @ignore_exceptions 358 @log_exceptions 359 async def update_channel_state(self, *, funding_outpoint: str, funding_txid: str, 360 funding_height: TxMinedInfo, closing_txid: str, 361 closing_height: TxMinedInfo, keep_watching: bool) -> None: 362 chan = self.lnworker.channel_by_txo(funding_outpoint) 363 if not chan: 364 return 365 chan.update_onchain_state(funding_txid=funding_txid, 366 funding_height=funding_height, 367 closing_txid=closing_txid, 368 closing_height=closing_height, 369 keep_watching=keep_watching) 370 await self.lnworker.on_channel_update(chan) 371 372 async def do_breach_remedy(self, funding_outpoint, closing_tx, spenders): 373 chan = self.lnworker.channel_by_txo(funding_outpoint) 374 if not chan: 375 return False 376 # detect who closed and set sweep_info 377 sweep_info_dict = chan.sweep_ctx(closing_tx) 378 keep_watching = False if sweep_info_dict else not self.is_deeply_mined(closing_tx.txid()) 379 self.logger.info(f'(chan {chan.get_id_for_log()}) sweep_info_dict {[x.name for x in sweep_info_dict.values()]}') 380 # create and broadcast transaction 381 for prevout, sweep_info in sweep_info_dict.items(): 382 name = sweep_info.name + ' ' + chan.get_id_for_log() 383 spender_txid = spenders.get(prevout) 384 if spender_txid is not None: 385 spender_tx = self.db.get_transaction(spender_txid) 386 if not spender_tx: 387 keep_watching = True 388 continue 389 e_htlc_tx = chan.maybe_sweep_revoked_htlc(closing_tx, spender_tx) 390 if e_htlc_tx: 391 spender2 = spenders.get(spender_txid+':0') 392 if spender2: 393 self.logger.info(f'(chan {chan.get_id_for_log()}) htlc is already spent {name}: {prevout}') 394 keep_watching |= not self.is_deeply_mined(spender2) 395 else: 396 self.logger.info(f'(chan {chan.get_id_for_log()}) trying to redeem htlc {name}: {prevout}') 397 await self.try_redeem(spender_txid+':0', e_htlc_tx, name) 398 keep_watching = True 399 else: 400 self.logger.info(f'(chan {chan.get_id_for_log()}) outpoint already spent {name}: {prevout}') 401 keep_watching |= not self.is_deeply_mined(spender_txid) 402 txin_idx = spender_tx.get_input_idx_that_spent_prevout(TxOutpoint.from_str(prevout)) 403 assert txin_idx is not None 404 spender_txin = spender_tx.inputs()[txin_idx] 405 chan.extract_preimage_from_htlc_txin(spender_txin) 406 else: 407 self.logger.info(f'(chan {chan.get_id_for_log()}) trying to redeem {name}: {prevout}') 408 await self.try_redeem(prevout, sweep_info, name) 409 keep_watching = True 410 return keep_watching 411 412 @log_exceptions 413 async def try_redeem(self, prevout: str, sweep_info: 'SweepInfo', name: str) -> None: 414 prev_txid, prev_index = prevout.split(':') 415 broadcast = True 416 if sweep_info.cltv_expiry: 417 local_height = self.network.get_local_height() 418 remaining = sweep_info.cltv_expiry - local_height 419 if remaining > 0: 420 self.logger.info('waiting for {}: CLTV ({} > {}), prevout {}' 421 .format(name, local_height, sweep_info.cltv_expiry, prevout)) 422 broadcast = False 423 if sweep_info.csv_delay: 424 prev_height = self.get_tx_height(prev_txid) 425 remaining = sweep_info.csv_delay - prev_height.conf 426 if remaining > 0: 427 self.logger.info('waiting for {}: CSV ({} >= {}), prevout: {}' 428 .format(name, prev_height.conf, sweep_info.csv_delay, prevout)) 429 broadcast = False 430 tx = sweep_info.gen_tx() 431 if tx is None: 432 self.logger.info(f'{name} could not claim output: {prevout}, dust') 433 self.lnworker.wallet.set_label(tx.txid(), name) 434 if broadcast: 435 await self.network.try_broadcasting(tx, name) 436 else: 437 # it's OK to add local transaction, the fee will be recomputed 438 try: 439 tx_was_added = self.lnworker.wallet.add_future_tx(tx, remaining) 440 except Exception as e: 441 self.logger.info(f'could not add future tx: {name}. prevout: {prevout} {str(e)}') 442 tx_was_added = False 443 if tx_was_added: 444 self.logger.info(f'added future tx: {name}. prevout: {prevout}') 445 util.trigger_callback('wallet_updated', self.lnworker.wallet)