tlnwatcher.py - electrum - Electrum Bitcoin wallet
HTML git clone https://git.parazyd.org/electrum
DIR Log
DIR Files
DIR Refs
DIR Submodules
---
tlnwatcher.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)