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