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 5from typing import NamedTuple, Iterable, TYPE_CHECKING 6import os 7import asyncio 8from enum import IntEnum, auto 9from typing import NamedTuple, Dict 10 11from . import util 12from .sql_db import SqlDB, sql 13from .wallet_db import WalletDB 14from .util import bh2u, bfh, log_exceptions, ignore_exceptions, TxMinedInfo, random_shuffled_copy 15from .address_synchronizer import AddressSynchronizer, TX_HEIGHT_LOCAL, TX_HEIGHT_UNCONF_PARENT, TX_HEIGHT_UNCONFIRMED 16from .transaction import Transaction, TxOutpoint 17 18if TYPE_CHECKING: 19 from .network import Network 20 from .lnsweep import SweepInfo 21 from .lnworker import LNWallet 22 23class 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 29class 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 37create_sweep_txs=""" 38CREATE TABLE IF NOT EXISTS sweep_txs ( 39funding_outpoint VARCHAR(34) NOT NULL, 40ctn INTEGER NOT NULL, 41prevout VARCHAR(34), 42tx VARCHAR 43)""" 44 45create_channel_info=""" 46CREATE TABLE IF NOT EXISTS channel_info ( 47outpoint VARCHAR(34) NOT NULL, 48address VARCHAR(32), 49PRIMARY KEY(outpoint) 50)""" 51 52 53class 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 135class 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 o.address is None: 238 continue 239 if not self.is_mine(o.address): 240 self.add_address(o.address) 241 elif n < 2: 242 r = self.inspect_tx_candidate(txid+':%d'%i, n+1) 243 result.update(r) 244 return result 245 246 def get_tx_mined_depth(self, txid: str): 247 if not txid: 248 return TxMinedDepth.FREE 249 tx_mined_depth = self.get_tx_height(txid) 250 height, conf = tx_mined_depth.height, tx_mined_depth.conf 251 if conf > 100: 252 return TxMinedDepth.DEEP 253 elif conf > 0: 254 return TxMinedDepth.SHALLOW 255 elif height in (TX_HEIGHT_UNCONFIRMED, TX_HEIGHT_UNCONF_PARENT): 256 return TxMinedDepth.MEMPOOL 257 elif height == TX_HEIGHT_LOCAL: 258 return TxMinedDepth.FREE 259 elif height > 0 and conf == 0: 260 # unverified but claimed to be mined 261 return TxMinedDepth.MEMPOOL 262 else: 263 raise NotImplementedError() 264 265 def is_deeply_mined(self, txid): 266 return self.get_tx_mined_depth(txid) == TxMinedDepth.DEEP 267 268 269class WatchTower(LNWatcher): 270 271 LOGGING_SHORTCUT = 'W' 272 273 def __init__(self, network): 274 LNWatcher.__init__(self, network) 275 self.network = network 276 self.sweepstore = SweepStore(os.path.join(self.network.config.path, "watchtower_db"), network) 277 # this maps funding_outpoints to ListenerItems, which have an event for when the watcher is done, 278 # and a queue for seeing which txs are being published 279 self.tx_progress = {} # type: Dict[str, ListenerItem] 280 281 def diagnostic_name(self): 282 return "local_tower" 283 284 async def start_watching(self): 285 # I need to watch the addresses from sweepstore 286 lst = await self.sweepstore.list_channels() 287 for outpoint, address in random_shuffled_copy(lst): 288 self.add_channel(outpoint, address) 289 290 async def do_breach_remedy(self, funding_outpoint, closing_tx, spenders): 291 keep_watching = False 292 for prevout, spender in spenders.items(): 293 if spender is not None: 294 keep_watching |= not self.is_deeply_mined(spender) 295 continue 296 sweep_txns = await self.sweepstore.get_sweep_tx(funding_outpoint, prevout) 297 for tx in sweep_txns: 298 await self.broadcast_or_log(funding_outpoint, tx) 299 keep_watching = True 300 return keep_watching 301 302 async def broadcast_or_log(self, funding_outpoint: str, tx: Transaction): 303 height = self.get_tx_height(tx.txid()).height 304 if height != TX_HEIGHT_LOCAL: 305 return 306 try: 307 txid = await self.network.broadcast_transaction(tx) 308 except Exception as e: 309 self.logger.info(f'broadcast failure: txid={tx.txid()}, funding_outpoint={funding_outpoint}: {repr(e)}') 310 else: 311 self.logger.info(f'broadcast success: txid={tx.txid()}, funding_outpoint={funding_outpoint}') 312 if funding_outpoint in self.tx_progress: 313 await self.tx_progress[funding_outpoint].tx_queue.put(tx) 314 return txid 315 316 def get_ctn(self, outpoint, addr): 317 async def f(): 318 return await self.sweepstore.get_ctn(outpoint, addr) 319 return self.network.run_from_another_thread(f()) 320 321 def get_num_tx(self, outpoint): 322 async def f(): 323 return await self.sweepstore.get_num_tx(outpoint) 324 return self.network.run_from_another_thread(f()) 325 326 def list_sweep_tx(self): 327 async def f(): 328 return await self.sweepstore.list_sweep_tx() 329 return self.network.run_from_another_thread(f()) 330 331 def list_channels(self): 332 async def f(): 333 return await self.sweepstore.list_channels() 334 return self.network.run_from_another_thread(f()) 335 336 async def unwatch_channel(self, address, funding_outpoint): 337 await super().unwatch_channel(address, funding_outpoint) 338 await self.sweepstore.remove_sweep_tx(funding_outpoint) 339 await self.sweepstore.remove_channel(funding_outpoint) 340 if funding_outpoint in self.tx_progress: 341 self.tx_progress[funding_outpoint].all_done.set() 342 343 async def update_channel_state(self, *args, **kwargs): 344 pass 345 346 347 348 349class LNWalletWatcher(LNWatcher): 350 351 def __init__(self, lnworker: 'LNWallet', network: 'Network'): 352 self.network = network 353 self.lnworker = lnworker 354 LNWatcher.__init__(self, network) 355 356 def diagnostic_name(self): 357 return f"{self.lnworker.wallet.diagnostic_name()}-LNW" 358 359 @ignore_exceptions 360 @log_exceptions 361 async def update_channel_state(self, *, funding_outpoint: str, funding_txid: str, 362 funding_height: TxMinedInfo, closing_txid: str, 363 closing_height: TxMinedInfo, keep_watching: bool) -> None: 364 chan = self.lnworker.channel_by_txo(funding_outpoint) 365 if not chan: 366 return 367 chan.update_onchain_state(funding_txid=funding_txid, 368 funding_height=funding_height, 369 closing_txid=closing_txid, 370 closing_height=closing_height, 371 keep_watching=keep_watching) 372 await self.lnworker.on_channel_update(chan) 373 374 async def do_breach_remedy(self, funding_outpoint, closing_tx, spenders): 375 chan = self.lnworker.channel_by_txo(funding_outpoint) 376 if not chan: 377 return False 378 chan_id_for_log = chan.get_id_for_log() 379 # detect who closed and set sweep_info 380 sweep_info_dict = chan.sweep_ctx(closing_tx) 381 keep_watching = False if sweep_info_dict else not self.is_deeply_mined(closing_tx.txid()) 382 # create and broadcast transaction 383 for prevout, sweep_info in sweep_info_dict.items(): 384 name = sweep_info.name + ' ' + chan.get_id_for_log() 385 spender_txid = spenders.get(prevout) 386 if spender_txid is not None: 387 spender_tx = self.db.get_transaction(spender_txid) 388 if not spender_tx: 389 keep_watching = True 390 continue 391 e_htlc_tx = chan.maybe_sweep_revoked_htlc(closing_tx, spender_tx) 392 if e_htlc_tx: 393 spender2 = spenders.get(spender_txid+':0') 394 if spender2: 395 keep_watching |= not self.is_deeply_mined(spender2) 396 else: 397 await self.try_redeem(spender_txid+':0', e_htlc_tx, chan_id_for_log, name) 398 keep_watching = True 399 else: 400 keep_watching |= not self.is_deeply_mined(spender_txid) 401 txin_idx = spender_tx.get_input_idx_that_spent_prevout(TxOutpoint.from_str(prevout)) 402 assert txin_idx is not None 403 spender_txin = spender_tx.inputs()[txin_idx] 404 chan.extract_preimage_from_htlc_txin(spender_txin) 405 else: 406 await self.try_redeem(prevout, sweep_info, chan_id_for_log, name) 407 keep_watching = True 408 return keep_watching 409 410 @log_exceptions 411 async def try_redeem(self, prevout: str, sweep_info: 'SweepInfo', chan_id_for_log: str, name: str) -> None: 412 prev_txid, prev_index = prevout.split(':') 413 broadcast = True 414 local_height = self.network.get_local_height() 415 if sweep_info.cltv_expiry: 416 wanted_height = sweep_info.cltv_expiry - local_height 417 if wanted_height - local_height > 0: 418 broadcast = False 419 reason = 'waiting for {}: CLTV ({} > {}), prevout {}'.format(name, local_height, sweep_info.cltv_expiry, prevout) 420 if sweep_info.csv_delay: 421 prev_height = self.get_tx_height(prev_txid) 422 wanted_height = sweep_info.csv_delay + prev_height.height - 1 423 if wanted_height - local_height > 0: 424 broadcast = False 425 reason = 'waiting for {}: CSV ({} >= {}), prevout: {}'.format(name, prev_height.conf, sweep_info.csv_delay, prevout) 426 tx = sweep_info.gen_tx() 427 if tx is None: 428 self.logger.info(f'{name} could not claim output: {prevout}, dust') 429 return 430 txid = tx.txid() 431 self.lnworker.wallet.set_label(txid, name) 432 if broadcast: 433 await self.network.try_broadcasting(tx, name) 434 else: 435 if txid in self.lnworker.wallet.future_tx: 436 return 437 self.logger.debug(f'(chan {chan_id_for_log}) trying to redeem {name}: {prevout}') 438 self.logger.info(reason) 439 # it's OK to add local transaction, the fee will be recomputed 440 try: 441 tx_was_added = self.lnworker.wallet.add_future_tx(tx, wanted_height) 442 except Exception as e: 443 self.logger.info(f'could not add future tx: {name}. prevout: {prevout} {str(e)}') 444 tx_was_added = False 445 if tx_was_added: 446 self.logger.info(f'added future tx: {name}. prevout: {prevout}') 447 util.trigger_callback('wallet_updated', self.lnworker.wallet) 448 449 def add_verified_tx(self, tx_hash: str, info: TxMinedInfo): 450 # this method is overloaded so that we have the GUI refreshed 451 # TODO: LNWatcher should not be an AddressSynchronizer, 452 # we should use the existing wallet instead, and results would be persisted 453 super().add_verified_tx(tx_hash, info) 454 tx_mined_status = self.get_tx_height(tx_hash) 455 util.trigger_callback('verified', self.lnworker.wallet, tx_hash, tx_mined_status) 456