commit 6a8e5d59549cbc0384686af2cbf416787e6e4494
parent 4268be90932dff31c94ef2182e8d59ff51872d43
Author: Janus <ysangkok@gmail.com>
Date: Mon, 28 May 2018 18:22:45 +0200
ln: restore functionality
Diffstat:
4 files changed, 103 insertions(+), 50 deletions(-)
diff --git a/electrum/commands.py b/electrum/commands.py
@@ -773,8 +773,8 @@ class Commands:
self.wallet.lnworker.reestablish_channel()
@command('wn')
- def lnpay():
- self.wallet.lnworker.pay()
+ def lnpay(self, invoice):
+ self.wallet.lnworker.pay(invoice)
@command('wn')
def lnreceive(self):
diff --git a/lib/lnbase.py b/lib/lnbase.py
@@ -39,10 +39,12 @@ from .lnrouter import new_onion_packet, OnionHopsDataSingle, OnionPerHop
from collections import namedtuple, defaultdict
-
class LightningError(Exception):
pass
+class LightningPeerConnectionClosed(LightningError):
+ pass
+
message_types = {}
def handlesingle(x, ma):
@@ -566,7 +568,8 @@ def is_synced(network):
class Peer(PrintError):
- def __init__(self, host, port, pubkey, privkey, network, channel_db, path_finder, request_initial_sync=False):
+ def __init__(self, host, port, pubkey, privkey, network, channel_db, path_finder, channel_state, handle_channel_reestablish, request_initial_sync=False):
+ self.handle_channel_reestablish = handle_channel_reestablish
self.update_add_htlc_event = asyncio.Event()
self.channel_update_event = asyncio.Event()
self.host = host
@@ -576,7 +579,6 @@ class Peer(PrintError):
self.network = network
self.channel_db = channel_db
self.path_finder = path_finder
- self.read_buffer = b''
self.ping_time = 0
self.futures = ["channel_accepted",
"funding_signed",
@@ -597,6 +599,8 @@ class Peer(PrintError):
self.initialized = asyncio.Future()
self.localfeatures = (0x08 if request_initial_sync else 0)
self.unfulfilled_htlcs = []
+ self.channel_state = channel_state
+ self.nodes = {}
def diagnostic_name(self):
return self.host
@@ -619,21 +623,22 @@ class Peer(PrintError):
async def read_message(self):
rn_l, rk_l = self.rn()
rn_m, rk_m = self.rn()
+ read_buffer = b''
while True:
s = await self.reader.read(1)
if not s:
- raise Exception('connection closed')
- self.read_buffer += s
- if len(self.read_buffer) < 18:
+ raise LightningPeerConnectionClosed()
+ read_buffer += s
+ if len(read_buffer) < 18:
continue
- lc = self.read_buffer[:18]
+ lc = read_buffer[:18]
l = aead_decrypt(rk_l, rn_l, b'', lc)
length = int.from_bytes(l, 'big')
offset = 18 + length + 16
- if len(self.read_buffer) < offset:
+ if len(read_buffer) < offset:
continue
- c = self.read_buffer[18:offset]
- self.read_buffer = self.read_buffer[offset:]
+ c = read_buffer[18:offset]
+ read_buffer = read_buffer[offset:]
msg = aead_decrypt(rk_m, rn_m, b'', c)
return msg
@@ -716,7 +721,7 @@ class Peer(PrintError):
if chan_id in self.channel_reestablish:
self.channel_reestablish[chan_id].set_result(payload)
else:
- print("Warning: received unknown channel_reestablish")
+ asyncio.run_coroutine_threadsafe(self.handle_channel_reestablish(chan_id, payload), self.network.asyncio_loop).result()
def on_accept_channel(self, payload):
temp_chan_id = payload["temporary_channel_id"]
@@ -906,6 +911,7 @@ class Peer(PrintError):
sig_64 = sign_and_get_sig_string(remote_ctx, local_config, remote_config)
funding_txid_bytes = bytes.fromhex(funding_txid)[::-1]
channel_id = int.from_bytes(funding_txid_bytes, 'big') ^ funding_index
+ self.channel_state[channel_id] = "OPENING"
self.send_message(gen_msg("funding_created",
temporary_channel_id=temp_channel_id,
funding_txid=funding_txid_bytes,
@@ -958,6 +964,7 @@ class Peer(PrintError):
return chan
async def reestablish_channel(self, chan):
+ assert chan.channel_id not in self.channel_state
await self.initialized
self.send_message(gen_msg("channel_reestablish",
@@ -984,9 +991,11 @@ class Peer(PrintError):
if channel_reestablish_msg["my_current_per_commitment_point"] != chan.remote_state.last_per_commitment_point:
raise Exception("Remote PCP mismatch")
- return chan
- async def on_funding_locked(self):
+ self.channel_state[chan.channel_id] = "OPEN"
+
+ async def funding_locked(self, chan):
+ channel_id = chan.channel_id
try:
short_channel_id = await self.local_funding_locked[channel_id]
finally:
@@ -1003,6 +1012,8 @@ class Peer(PrintError):
del self.remote_funding_locked[channel_id]
self.print_error('Done waiting for remote_funding_locked', remote_funding_locked_msg)
+ self.channel_state[chan.channel_id] = "OPEN"
+
return chan._replace(short_channel_id=short_channel_id, remote_state=chan.remote_state._replace(next_per_commitment_point=remote_funding_locked_msg["next_per_commitment_point"]))
def on_update_fail_htlc(self, payload):
@@ -1025,11 +1036,16 @@ class Peer(PrintError):
)
)
return last_secret, this_point, next_point
+ assert self.channel_state[chan.channel_id] == "OPEN"
their_revstore = chan.remote_state.revocation_store
- await asyncio.sleep(1)
while not is_synced(wallet.network):
await asyncio.sleep(1)
print("sleeping more")
+
+ if chan.channel_id in self.commitment_signed:
+ print("too many commitments signed")
+ del self.commitment_signed[chan.channel_id]
+
height = wallet.get_local_height()
assert amount_msat > 0, "amount_msat is not greater zero"
@@ -1175,6 +1191,8 @@ class Peer(PrintError):
)
return last_secret, this_point, next_point
+ assert self.channel_state[chan.channel_id] == "OPEN"
+
their_revstore = chan.remote_state.revocation_store
channel_id = chan.channel_id
@@ -1183,8 +1201,15 @@ class Peer(PrintError):
finally:
del self.commitment_signed[channel_id]
- assert len(self.unfulfilled_htlcs) == 1
- htlc = self.unfulfilled_htlcs.pop()
+ if int.from_bytes(commitment_signed_msg["num_htlcs"], "big") < 1:
+ while len(self.unfulfilled_htlcs) < 1:
+ print("waiting for add_update_htlc")
+ await asyncio.sleep(1)
+ else:
+ print("commitment signed message had htlcs")
+ assert len(self.unfulfilled_htlcs) == 1
+
+ htlc = self.unfulfilled_htlcs.pop(0)
htlc_id = int.from_bytes(htlc["id"], 'big')
assert htlc_id == chan.remote_state.next_htlc_id, (htlc_id, chan.remote_state.next_htlc_id)
cltv_expiry = int.from_bytes(htlc["cltv_expiry"], 'big')
diff --git a/lib/lnworker.py b/lib/lnworker.py
@@ -10,6 +10,7 @@ import binascii
import asyncio
+from . import constants
from .bitcoin import sha256, COIN
from .util import bh2u, bfh
from .constants import set_testnet, set_simnet
@@ -17,7 +18,7 @@ from .simple_config import SimpleConfig
from .network import Network
from .storage import WalletStorage
from .wallet import Wallet
-from .lnbase import Peer, Outpoint, ChannelConfig, LocalState, RemoteState, Keypair, OnlyPubkeyKeypair, OpenChannel, ChannelConstraints, RevocationStore, aiosafe
+from .lnbase import Peer, Outpoint, ChannelConfig, LocalState, RemoteState, Keypair, OnlyPubkeyKeypair, OpenChannel, ChannelConstraints, RevocationStore, aiosafe, calc_short_channel_id, privkey_to_pubkey
from .lightning_payencode.lnaddr import lnencode, LnAddr, lndecode
from . import lnrouter
@@ -98,92 +99,119 @@ class LNWorker:
self.channel_db = lnrouter.ChannelDB()
self.path_finder = lnrouter.LNPathFinder(self.channel_db)
- self.channels = wallet.storage.get("channels", {})
+ self.channels = [reconstruct_namedtuples(x) for x in wallet.storage.get("channels", {})]
peer_list = network.config.get('lightning_peers', node_list)
+ self.channel_state = {}
for host, port, pubkey in peer_list:
- self.add_peer(host, port, pubkey)
+ self.add_peer(host, int(port), pubkey)
# wait until we see confirmations
self.network.register_callback(self.on_network_update, ['updated', 'verified']) # thread safe
self.on_network_update('updated') # shortcut (don't block) if funding tx locked and verified
def add_peer(self, host, port, pubkey):
peer = Peer(host, int(port), binascii.unhexlify(pubkey), self.privkey,
- self.network, self.channel_db, self.path_finder)
+ self.network, self.channel_db, self.path_finder, self.channel_state, self.handle_channel_reestablish)
self.network.futures.append(asyncio.run_coroutine_threadsafe(peer.main_loop(), asyncio.get_event_loop()))
- self.peers[pubkey] = peer
+ self.peers[bfh(pubkey)] = peer
+
+ async def handle_channel_reestablish(self, chan_id, payload):
+ chans = [x for x in self.channels if x.channel_id == chan_id ]
+ chan = chans[0]
+ await self.peers[chan.node_id].reestablish_channel(chan)
def save_channel(self, openchannel):
- dumped = serialize_channels([openchannel])
+ self.channels = [openchannel] # TODO multiple channels
+ dumped = serialize_channels(self.channels)
self.wallet.storage.put("channels", dumped)
self.wallet.storage.write()
def on_network_update(self, event, *args):
for chan in self.channels:
peer = self.peers[chan.node_id]
- conf = wallet.get_tx_height(chan.funding_outpoint.txid)[1]
+ conf = self.wallet.get_tx_height(chan.funding_outpoint.txid)[1]
if conf >= chan.constraints.funding_txn_minimum_depth:
- block_height, tx_pos = wallet.get_txpos(chan.funding_outpoint.txid)
+ block_height, tx_pos = self.wallet.get_txpos(chan.funding_outpoint.txid)
if tx_pos == -1:
self.print_error('funding tx is not yet SPV verified.. but there are '
'already enough confirmations (currently {})'.format(conf))
return
- asyncio.run_coroutine_threadsafe(self.set_local_funding_locked_result(peer, chan, block_height, txpos), asyncio.get_event_loop())
+ if chan.channel_id not in self.channel_state or self.channel_state[chan.channel_id] != "OPENING":
+ return
+ asyncio.run_coroutine_threadsafe(self.set_local_funding_locked_result(peer, chan, block_height, tx_pos), asyncio.get_event_loop())
- async def set_local_funding_locked_result(self, peer, chan, block_height, txpos):
+ # aiosafe because we don't wait for result
+ @aiosafe
+ async def set_local_funding_locked_result(self, peer, chan, block_height, tx_pos):
channel_id = chan.channel_id
+ short_channel_id = calc_short_channel_id(block_height, tx_pos, chan.funding_outpoint.output_index)
try:
peer.local_funding_locked[channel_id].set_result(short_channel_id)
except (asyncio.InvalidStateError, KeyError) as e:
# FIXME race condition if updates come in quickly, set_result might be called multiple times
# or self.local_funding_locked[channel_id] might be deleted already
self.print_error('local_funding_locked.set_result error for channel {}: {}'.format(channel_id, e))
- short_channel_id = calc_short_channel_id(block_height, tx_pos, chan.funding_outpoint.output_index)
- openchannel = await peer.on_funding_locked(openingchannel, self.wallet)
+ openchannel = await peer.funding_locked(chan)
self.save_channel(openchannel)
+ print("CHANNEL OPENING COMPLETED")
- @aiosafe
+ # not aiosafe because we call .result() which will propagate an exception
async def _open_channel_coroutine(self, node_id, amount, push_msat, password):
- peer = self.peers[node_id]
+ peer = self.peers[bfh(node_id)]
openingchannel = await peer.channel_establishment_flow(self.wallet, self.config, password, amount, push_msat, temp_channel_id=os.urandom(32))
self.save_channel(openingchannel)
- def open_channel(self, node_id, local_amt, push_amt, emit_function, pw):
+ def open_channel(self, node_id, local_amt, push_amt, pw):
coro = self._open_channel_coroutine(node_id, local_amt, push_amt, None if pw == "" else pw)
- asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop)
-
+ return asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop).result()
#chan = fut.result()
# https://api.lightning.community/#listchannels
#std_chan = {"chan_id": chan.channel_id}
#emit_function({"channels": [std_chan]})
+ def get_paid(self):
+ coro = self._get_paid_coroutine()
+ return asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop).result()
+
+ def pay(self, invoice):
+ coro = self._pay_coroutine(invoice)
+ return asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop).result()
+
def list_channels(self):
- return self.channels
+ return serialize_channels(self.channels)
- @aiosafe
- async def reestablish_channel(self):
+ def reestablish_channels(self):
+ coro = self._reestablish_channels_coroutine()
+ return asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop).result()
+
+ # not aiosafe because we call .result() which will propagate an exception
+ async def _reestablish_channels_coroutine(self):
if self.channels is None or len(self.channels) < 1:
raise Exception("Can't reestablish: No channel saved")
- openchannel = self.channels[0]
- openchannel = reconstruct_namedtuples(openchannel)
- openchannel = await peer.reestablish_channel(openchannel)
- self.save_channel(openchannel)
+ peer = self.peers[self.channels[0].node_id]
+ await peer.reestablish_channel(self.channels[0])
- @aiosafe
- async def pay(self):
- addr = lndecode(sys.argv[6], expected_hrp="sb" if sys.argv[2] == "simnet" else "tb")
+ # not aiosafe because we call .result() which will propagate an exception
+ async def _pay_coroutine(self, invoice):
+ openchannel = self.channels[0]
+ addr = lndecode(invoice, expected_hrp=constants.net.SEGWIT_HRP)
payment_hash = addr.paymenthash
pubkey = addr.pubkey.serialize()
msat_amt = int(addr.amount * COIN * 1000)
- openchannel = await peer.pay(wallet, openchannel, msat_amt, payment_hash, pubkey, addr.min_final_cltv_expiry)
+ peer = self.peers[openchannel.node_id]
+ openchannel = await peer.pay(self.wallet, openchannel, msat_amt, payment_hash, pubkey, addr.min_final_cltv_expiry)
self.save_channel(openchannel)
- @aiosafe
- async def get_paid(self):
+ # not aiosafe because we call .result() which will propagate an exception
+ async def _get_paid_coroutine(self):
+ openchannel = self.channels[0]
payment_preimage = os.urandom(32)
RHASH = sha256(payment_preimage)
expected_received_sat = 200000
expected_received_msat = expected_received_sat * 1000
- pay_req = lnencode(LnAddr(RHASH, amount=1/Decimal(COIN)*expected_received_sat, tags=[('d', 'one cup of coffee')]), peer.privkey[:32])
+ peer = self.peers[openchannel.node_id]
+ pay_req = lnencode(LnAddr(RHASH, amount=1/Decimal(COIN)*expected_received_sat, tags=[('d', 'one cup of coffee')]), peer.privkey)
+ decoded = lndecode(pay_req, expected_hrp="sb")
+ assert decoded.pubkey.serialize() == privkey_to_pubkey(self.privkey)
print("payment request", pay_req)
openchannel = await peer.receive_commitment_revoke_ack(openchannel, expected_received_msat, payment_preimage)
self.save_channel(openchannel)
diff --git a/lib/tests/test_lnbase.py b/lib/tests/test_lnbase.py
@@ -256,7 +256,7 @@ class Test_LNBase(unittest.TestCase):
def test_find_path_for_payment(self):
channel_db = lnrouter.ChannelDB()
path_finder = lnrouter.LNPathFinder(channel_db)
- p = Peer('', 0, 'a', bitcoin.sha256('privkeyseed'), None, channel_db, path_finder)
+ p = Peer('', 0, 'a', bitcoin.sha256('privkeyseed'), None, channel_db, path_finder, {}, lambda x, y: None)
p.on_channel_announcement({'node_id_1': b'b', 'node_id_2': b'c', 'short_channel_id': bfh('0000000000000001')})
p.on_channel_announcement({'node_id_1': b'b', 'node_id_2': b'e', 'short_channel_id': bfh('0000000000000002')})
p.on_channel_announcement({'node_id_1': b'a', 'node_id_2': b'b', 'short_channel_id': bfh('0000000000000003')})
