From df93cbdcee1ebe6d3cdc7d091d6ac74161c20da8 Mon Sep 17 00:00:00 2001 From: fusion44 Date: Mon, 3 Oct 2022 10:44:34 +0200 Subject: [PATCH] refactor: lightning system decoupling --- app/repositories/lightning.py | 56 +- app/repositories/ln_impl/cln_grpc.py | 1430 +++++++-------- app/repositories/ln_impl/ln_base.py | 133 ++ app/repositories/ln_impl/lnd_grpc.py | 1536 ++++++++--------- .../ln_impl/specializations/cln_grpc_blitz.py | 492 +++--- 5 files changed, 1858 insertions(+), 1789 deletions(-) create mode 100644 app/repositories/ln_impl/ln_base.py diff --git a/app/repositories/lightning.py b/app/repositories/lightning.py index da102f2..ba64659 100644 --- a/app/repositories/lightning.py +++ b/app/repositories/lightning.py @@ -28,11 +28,13 @@ PLATFORM = config("platform", cast=str) ln_node = config("ln_node") if ln_node == "lnd_grpc": - import app.repositories.ln_impl.lnd_grpc as ln + from app.repositories.ln_impl.lnd_grpc import LnNodeLNDgRPC as LnNode elif ln_node == "cln_grpc" and PLATFORM != APIPlatform.RASPIBLITZ: - import app.repositories.ln_impl.cln_grpc as ln + from app.repositories.ln_impl.cln_grpc import LnNodeCLNgRPC as LnNode elif ln_node == "cln_grpc" and PLATFORM == APIPlatform.RASPIBLITZ: - import app.repositories.ln_impl.specializations.cln_grpc_blitz as ln + from app.repositories.ln_impl.specializations.cln_grpc_blitz import ( + LnNodeCLNgRPCBlitz as LnNode, + ) elif ln_node == "none": logging.info(f"lightning was explicitly turned off") elif ln_node == "": @@ -54,31 +56,33 @@ FWD_GATHER_INTERVAL = config("forwards_gather_interval", default=2.0, cast=float if FWD_GATHER_INTERVAL < 0.3: raise RuntimeError("forwards_gather_interval cannot be less than 0.3 seconds") +ln = LnNode() + async def initialize_ln_repo() -> AsyncGenerator[InitLnRepoUpdate, None]: - async for u in ln.initialize_impl(): + async for u in ln.initialize(): yield u async def get_ln_info_lite() -> LightningInfoLite: - ln_info = await ln.get_ln_info_impl() + ln_info = await ln.get_ln_info() return LightningInfoLite.from_lninfo(ln_info) async def get_wallet_balance(): - return await ln.get_wallet_balance_impl() + return await ln.get_wallet_balance() async def list_all_tx( successful_only: bool, index_offset: int, max_tx: int, reversed: bool ) -> List[GenericTx]: - return await ln.list_all_tx_impl(successful_only, index_offset, max_tx, reversed) + return await ln.list_all_tx(successful_only, index_offset, max_tx, reversed) async def list_invoices( pending_only: bool, index_offset: int, num_max_invoices: int, reversed: bool ) -> List[Invoice]: - return await ln.list_invoices_impl( + return await ln.list_invoices( pending_only, index_offset, num_max_invoices, @@ -87,13 +91,13 @@ async def list_invoices( async def list_on_chain_tx() -> List[OnChainTransaction]: - return await ln.list_on_chain_tx_impl() + return await ln.list_on_chain_tx() async def list_payments( include_incomplete: bool, index_offset: int, max_payments: int, reversed: bool ) -> List[Payment]: - return await ln.list_payments_impl( + return await ln.list_payments( include_incomplete, index_offset, max_payments, reversed ) @@ -101,19 +105,19 @@ async def list_payments( async def add_invoice( value_msat: int, memo: str = "", expiry: int = 3600, is_keysend: bool = False ) -> Invoice: - return await ln.add_invoice_impl(memo, value_msat, expiry, is_keysend) + return await ln.add_invoice(memo, value_msat, expiry, is_keysend) async def decode_pay_request(pay_req: str) -> PaymentRequest: - return await ln.decode_pay_request_impl(pay_req) + return await ln.decode_pay_request(pay_req) async def new_address(input: NewAddressInput) -> str: - return await ln.new_address_impl(input) + return await ln.new_address(input) async def send_coins(input: SendCoinsInput) -> SendCoinsResponse: - res = await ln.send_coins_impl(input) + res = await ln.send_coins(input) _schedule_wallet_balance_update() return res @@ -124,9 +128,7 @@ async def send_payment( fee_limit_msat: int, amount_msat: Optional[int] = None, ) -> Payment: - res = await ln.send_payment_impl( - pay_req, timeout_seconds, fee_limit_msat, amount_msat - ) + res = await ln.send_payment(pay_req, timeout_seconds, fee_limit_msat, amount_msat) _schedule_wallet_balance_update() return res @@ -147,41 +149,41 @@ async def channel_open( if not "@" in node_URI: raise ValueError("node_URI must contain @ with node physical address") - res = await ln.channel_open_impl(local_funding_amount, node_URI, target_confs) + res = await ln.channel_open(local_funding_amount, node_URI, target_confs) return res async def channel_list() -> List[Channel]: - res = await ln.channel_list_impl() + res = await ln.channel_list() return res async def channel_close(channel_id: int, force_close: bool) -> str: - res = await ln.channel_close_impl(channel_id, force_close) + res = await ln.channel_close(channel_id, force_close) return res async def get_ln_info() -> LnInfo: - ln_info = await ln.get_ln_info_impl() + ln_info = await ln.get_ln_info() if PLATFORM == APIPlatform.RASPIBLITZ: ln_info.identity_uri = await redis_get("ln_default_address") return ln_info async def unlock_wallet(password: str) -> bool: - res = await ln.unlock_wallet_impl(password) + res = await ln.unlock_wallet(password) return res async def get_fee_revenue() -> FeeRevenue: - return await ln.get_fee_revenue_impl() + return await ln.get_fee_revenue() async def register_lightning_listener(): """ Registers all lightning listeners - By calling get_ln_info_impl() once, we ensure that wallet is unlocked. + By calling get_ln_info() once, we ensure that wallet is unlocked. Implementation will throw HTTPException with status_code 423_LOCKED if otherwise. It is the task of the caller to call register_lightning_listener() again """ @@ -194,7 +196,7 @@ async def register_lightning_listener(): ) return - await ln.get_ln_info_impl() + await ln.get_ln_info() loop = asyncio.get_event_loop() loop.create_task(_handle_info_listener()) @@ -208,7 +210,7 @@ async def _handle_info_listener(): last_info = None last_info_lite = None while True: - info = await ln.get_ln_info_impl() + info = await ln.get_ln_info() if last_info != info: await broadcast_sse_msg(SSE.LN_INFO, info.dict()) @@ -270,7 +272,7 @@ def _schedule_wallet_balance_update(): global _wallet_balance_update_scheduled _wallet_balance_update_scheduled = True await asyncio.sleep(1.1) - wb = await ln.get_wallet_balance_impl() + wb = await ln.get_wallet_balance() if _CACHE["wallet_balance"] != wb: await broadcast_sse_msg(SSE.WALLET_BALANCE, wb.dict()) _CACHE["wallet_balance"] = wb diff --git a/app/repositories/ln_impl/cln_grpc.py b/app/repositories/ln_impl/cln_grpc.py index b8c6972..5a5ae6c 100644 --- a/app/repositories/ln_impl/cln_grpc.py +++ b/app/repositories/ln_impl/cln_grpc.py @@ -35,29 +35,9 @@ from app.models.lightning import ( TxStatus, WalletBalance, ) +from app.repositories.ln_impl.ln_base import LightningNodeBase from app.repositories.utils.bitcoin import bitcoin_rpc_async -_cln_grpc_cert = bytes.fromhex( - config_get_hex_str(config("cln_grpc_cert"), name="cln_grpc_cert") -) -_cln_grpc_key = bytes.fromhex( - config_get_hex_str(config("cln_grpc_key"), name="cln_grpc_key") -) -_cln_grpc_ca = bytes.fromhex( - config_get_hex_str(config("cln_grpc_ca"), name="cln_grpc_ca") -) -_cln_grpc_url = config("cln_grpc_ip") + ":" + config("cln_grpc_port") -_creds = grpc.ssl_channel_credentials( - root_certificates=_cln_grpc_ca, - private_key=_cln_grpc_key, - certificate_chain=_cln_grpc_cert, -) -_opts = (("grpc.ssl_target_name_override", "cln"),) -_channel = None -_cln_stub: clnrpc.NodeStub = None - -_initialized = False - async def _make_local_call(cmd: str): # FIXME: this is a hack because some of the commands are not exposed @@ -103,763 +83,785 @@ async def _make_local_call(cmd: str): return stdout, stderr -def get_implementation_name() -> str: - return "CLN_GRPC" +class LnNodeCLNgRPC(LightningNodeBase): + _initialized = False + _channel = None + _cln_stub: clnrpc.NodeStub = None + # Decoding the payment request take a long time, + # hence we build a simple cache here. + _memo_cache = {} + _block_cache = {} + def get_implementation_name(self) -> str: + return "CLN_GRPC" -async def initialize_impl() -> AsyncGenerator[InitLnRepoUpdate, None]: - logging.debug("CLN_GRPC: Unable to connect to CLN daemon, waiting...") - - global _initialized - global _channel - global _cln_stub - - if _initialized: - logging.warning( - "CLN_GRPC: Connection already initialized. This function must not be called twice." - ) - yield InitLnRepoUpdate(state=LnInitState.DONE) - - while not _initialized: - try: - if _channel is None: - _channel = grpc.aio.secure_channel(_cln_grpc_url, _creds, options=_opts) - _cln_stub = clnrpc.NodeStub(_channel) - - await _cln_stub.Getinfo(ln.GetinfoRequest()) - _initialized = True + async def initialize(self) -> AsyncGenerator[InitLnRepoUpdate, None]: + logging.debug("CLN_GRPC: Unable to connect to CLN daemon, waiting...") + if self._initialized: + logging.warning( + "CLN_GRPC: Connection already initialized. This function must not be called twice." + ) yield InitLnRepoUpdate(state=LnInitState.DONE) - except grpc.aio._call.AioRpcError as error: - details = error.details() - logging.debug(f"CLN_GRPC: Waiting for CLN daemon... Details {details}") - if "failed to connect to all addresses" in details: - yield InitLnRepoUpdate( - state=LnInitState.OFFLINE, - msg="Unable to connect to CLN daemon, waiting...", - ) - - await _channel.close() - _channel = _cln_stub = None - else: - logging.error(f"CLN_GRPC: Unknown error: {details}") - raise - - await asyncio.sleep(2) - - logging.info("CLN_GRPC: Initialization complete.") - - -async def get_wallet_balance_impl() -> WalletBalance: - logging.debug("CLN_GRPC: get_wallet_balance_impl() ") - - req = ln.ListfundsRequest() - res = await _cln_stub.ListFunds(req) - onchain_confirmed = onchain_unconfirmed = onchain_total = 0 - - for o in res.outputs: - sat = o.amount_msat.msat / 1000 - onchain_total += sat - if o.status == 0: - onchain_unconfirmed += sat - elif o.status == 1: - onchain_confirmed += sat - # 2 is spent => ignore - - chan_local = chan_remote = chan_pending_local = chan_pending_remote = 0 - for c in res.channels: - our_msat = c.our_amount_msat.msat - their_msat = c.amount_msat.msat - our_msat - - if c.state == 2: # ChanneldNormal - chan_local += our_msat - chan_remote += their_msat - else: - # treat everything else as pending for now - chan_pending_local += our_msat - chan_pending_remote += their_msat - - return WalletBalance( - onchain_confirmed_balance=onchain_confirmed, - onchain_total_balance=onchain_total, - onchain_unconfirmed_balance=onchain_unconfirmed, - channel_local_balance=chan_local, - channel_remote_balance=chan_remote, - # TODO: find out how to get these values with CLN - channel_unsettled_local_balance=0, - channel_unsettled_remote_balance=0, - channel_pending_open_local_balance=chan_pending_local, - channel_pending_open_remote_balance=chan_pending_remote, - ) - - -# Decoding the payment request take a long time, -# hence we build a simple cache here. -memo_cache = {} -block_cache = {} - - -async def _get_block_time(block_height: int) -> tuple: - logging.debug(f"CLN_GRPC: _get_block_time(block_height={block_height}) ") - - if block_height is None or block_height < 0: - raise ValueError("block_height cannot be None or negative") - - if block_height in block_cache: - return block_cache[block_height] - - res = await bitcoin_rpc_async("getblockstats", params=[block_height]) - hash = res["result"]["blockhash"] - block = await bitcoin_rpc_async("getblock", params=[hash]) - block_cache[block_height] = (block["result"]["time"], block["result"]["mediantime"]) - return block_cache[block_height] - - -# Decoding the payment request take a long time, -# hence we build a simple cache here. -memo_cache = {} - - -async def list_all_tx_impl( - successful_only: bool, index_offset: int, max_tx: int, reversed: bool -) -> List[GenericTx]: - logging.debug( - f"CLN_GRPC: list_all_tx_impl(successful_only={successful_only}, index_offset={index_offset}, max_tx={max_tx}, reversed={reversed})" - ) - - list_invoice_req = ln.ListinvoicesRequest() - list_payments_req = ln.ListpaysRequest() - - try: - res = await asyncio.gather( - *[ - _cln_stub.ListInvoices(list_invoice_req), - list_on_chain_tx_impl(), - _cln_stub.ListPays(list_payments_req), - get_ln_info_impl(), - ] + cln_grpc_cert = bytes.fromhex( + config_get_hex_str(config("cln_grpc_cert"), name="cln_grpc_cert") + ) + cln_grpc_key = bytes.fromhex( + config_get_hex_str(config("cln_grpc_key"), name="cln_grpc_key") + ) + cln_grpc_ca = bytes.fromhex( + config_get_hex_str(config("cln_grpc_ca"), name="cln_grpc_ca") + ) + cln_grpc_url = config("cln_grpc_ip") + ":" + config("cln_grpc_port") + self.creds = grpc.ssl_channel_credentials( + root_certificates=cln_grpc_ca, + private_key=cln_grpc_key, + certificate_chain=cln_grpc_cert, ) - tx = [] - for invoice in res[0].invoices: - i = GenericTx.from_cln_grpc_invoice(invoice) - if successful_only and i.status == TxStatus.SUCCEEDED: - tx.append(i) - continue - tx.append(i) - for transaction in res[1]: - t = GenericTx.from_cln_grpc_onchain_tx(transaction, res[3].block_height) - if successful_only and t.status == TxStatus.SUCCEEDED: - tx.append(t) - continue + opts = (("grpc.ssl_target_name_override", "cln"),) - tx.append(t) + while not self._initialized: + try: + if self._channel is None: + self._channel = grpc.aio.secure_channel( + cln_grpc_url, self.creds, options=opts + ) + self._cln_stub = clnrpc.NodeStub(self._channel) - for pay in res[2].pays: - comment = "" + await self._cln_stub.Getinfo(ln.GetinfoRequest()) + self._initialized = True + yield InitLnRepoUpdate(state=LnInitState.DONE) + except grpc.aio._call.AioRpcError as error: + details = error.details() + logging.debug(f"CLN_GRPC: Waiting for CLN daemon... Details {details}") - if pay.bolt11 is not None and len(pay.bolt11) > 0: - if pay.bolt11 in memo_cache: - comment = memo_cache[pay.bolt11] + if "failed to connect to all addresses" in details: + yield InitLnRepoUpdate( + state=LnInitState.OFFLINE, + msg="Unable to connect to CLN daemon, waiting...", + ) + + await self._channel.close() + self._channel = self.cln_stub = None else: - pr = await decode_pay_request_impl(pay.bolt11) - comment = pr.description - memo_cache[pay.bolt11] = pr.description + logging.error(f"CLN_GRPC: Unknown error: {details}") + raise - p = GenericTx.from_cln_grpc_payment(pay, comment) + await asyncio.sleep(2) + + logging.info("CLN_GRPC: Initialization complete.") + + async def get_wallet_balance(self) -> WalletBalance: + logging.debug("CLN_GRPC: get_wallet_balance() ") + + req = ln.ListfundsRequest() + res = await self._cln_stub.ListFunds(req) + onchain_confirmed = onchain_unconfirmed = onchain_total = 0 + + for o in res.outputs: + sat = o.amount_msat.msat / 1000 + onchain_total += sat + if o.status == 0: + onchain_unconfirmed += sat + elif o.status == 1: + onchain_confirmed += sat + # 2 is spent => ignore + + chan_local = chan_remote = chan_pending_local = chan_pending_remote = 0 + for c in res.channels: + our_msat = c.our_amount_msat.msat + their_msat = c.amount_msat.msat - our_msat + + if c.state == 2: # ChanneldNormal + chan_local += our_msat + chan_remote += their_msat + else: + # treat everything else as pending for now + chan_pending_local += our_msat + chan_pending_remote += their_msat + + return WalletBalance( + onchain_confirmed_balance=onchain_confirmed, + onchain_total_balance=onchain_total, + onchain_unconfirmed_balance=onchain_unconfirmed, + channel_local_balance=chan_local, + channel_remote_balance=chan_remote, + # TODO: find out how to get these values with CLN + channel_unsettled_local_balance=0, + channel_unsettled_remote_balance=0, + channel_pending_open_local_balance=chan_pending_local, + channel_pending_open_remote_balance=chan_pending_remote, + ) + + async def _get_block_time(self, block_height: int) -> tuple: + logging.debug(f"CLN_GRPC: _get_block_time(block_height={block_height}) ") + + if block_height is None or block_height < 0: + raise ValueError("block_height cannot be None or negative") + + if block_height in self._block_cache: + return self._block_cache[block_height] + + res = await bitcoin_rpc_async("getblockstats", params=[block_height]) + hash = res["result"]["blockhash"] + block = await bitcoin_rpc_async("getblock", params=[hash]) + self._block_cache[block_height] = ( + block["result"]["time"], + block["result"]["mediantime"], + ) + return self._block_cache[block_height] + + async def list_all_tx( + self, successful_only: bool, index_offset: int, max_tx: int, reversed: bool + ) -> List[GenericTx]: + logging.debug( + f"CLN_GRPC: list_all_tx(successful_only={successful_only}, index_offset={index_offset}, max_tx={max_tx}, reversed={reversed})" + ) + + list_invoice_req = ln.ListinvoicesRequest() + list_payments_req = ln.ListpaysRequest() + + try: + res = await asyncio.gather( + *[ + self._cln_stub.ListInvoices(list_invoice_req), + self.list_on_chain_tx(), + self._cln_stub.ListPays(list_payments_req), + self.get_ln_info(), + ] + ) + tx = [] + for invoice in res[0].invoices: + i = GenericTx.from_cln_grpc_invoice(invoice) + if successful_only and i.status == TxStatus.SUCCEEDED: + tx.append(i) + continue + tx.append(i) + + for transaction in res[1]: + t = GenericTx.from_cln_grpc_onchain_tx(transaction, res[3].block_height) + if successful_only and t.status == TxStatus.SUCCEEDED: + tx.append(t) + continue + + tx.append(t) + + for pay in res[2].pays: + comment = "" + + if pay.bolt11 is not None and len(pay.bolt11) > 0: + if pay.bolt11 in self._memo_cache: + comment = self._memo_cache[pay.bolt11] + else: + pr = await self.decode_pay_request(pay.bolt11) + comment = pr.description + self._memo_cache[pay.bolt11] = pr.description + + p = GenericTx.from_cln_grpc_payment(pay, comment) + + if successful_only and p.status == TxStatus.SUCCEEDED: + tx.append(p) + continue - if successful_only and p.status == TxStatus.SUCCEEDED: tx.append(p) - continue - tx.append(p) + def sortKey(e: GenericTx): + return e.time_stamp - def sortKey(e: GenericTx): - return e.time_stamp + tx.sort(key=sortKey) - tx.sort(key=sortKey) + if reversed: + tx.reverse() + + l = len(tx) + for invoice in range(l): + tx[invoice].index = invoice + + if max_tx == 0: + max_tx = l + + return tx[index_offset : index_offset + max_tx] + except grpc.aio._call.AioRpcError as error: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def list_invoices( + self, + pending_only: bool, + index_offset: int, + num_max_invoices: int, + reversed: bool, + ) -> List[Invoice]: + logging.debug("CLN_GRPC: list_invoices() ") + + req = ln.ListinvoicesRequest() + res = await self._cln_stub.ListInvoices(req) + + tx = [] + for i in res.invoices: + if pending_only: + if i.status == 0: + tx.append(Invoice.from_cln_grpc(i)) + else: + tx.append(Invoice.from_cln_grpc(i)) if reversed: tx.reverse() - l = len(tx) - for invoice in range(l): - tx[invoice].index = invoice + if num_max_invoices == 0 or num_max_invoices is None: + return tx - if max_tx == 0: - max_tx = l + return tx[index_offset : index_offset + num_max_invoices] - return tx[index_offset : index_offset + max_tx] - except grpc.aio._call.AioRpcError as error: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) + async def list_on_chain_tx(self) -> List[OnChainTransaction]: + logging.debug("CLN_GRPC: list_on_chain_tx() ") + # Make a temporary copy of the file to avoid locking the db. + # CLN might want to write while we read. + info = await self.get_ln_info() -async def list_invoices_impl( - pending_only: bool, index_offset: int, num_max_invoices: int, reversed: bool -) -> List[Invoice]: - logging.debug("CLN_GRPC: list_invoices_impl() ") + # FIXME(#87): Once Core Lightnings accountability plugin is available + src = "/home/bitcoin/.lightning/bitcoin/lightningd.sqlite3" + dest = "/tmp/lightningd.sqlite3" + shutil.copyfile(src, dest) - req = ln.ListinvoicesRequest() - res = await _cln_stub.ListInvoices(req) + conn = sqlite3.connect(dest, uri=True) + cur = conn.execute("select * from outputs") + res = cur.fetchall() + conn.close() - tx = [] - for i in res.invoices: - if pending_only: - if i.status == 0: - tx.append(Invoice.from_cln_grpc(i)) - else: - tx.append(Invoice.from_cln_grpc(i)) + txs = [] + for o in res: + prev_out_tx = o[0].hex() + amount = o[2] + conf_block = o[9] + spent_block = o[10] + conf_time = (await self._get_block_time(conf_block))[0] + tx_hash = f"prev_out_tx {prev_out_tx}" - if reversed: - tx.reverse() + confs = info.block_height - conf_block + if confs < 0: + confs = 0 + logging.error( + f"Got negative confirmation count of for {tx_hash}\nCalc:{info.block_height} - {conf_block} = {confs}" + ) - if num_max_invoices == 0 or num_max_invoices is None: - return tx - - return tx[index_offset : index_offset + num_max_invoices] - - -async def list_on_chain_tx_impl() -> List[OnChainTransaction]: - logging.debug("CLN_GRPC: list_on_chain_tx_impl() ") - - # Make a temporary copy of the file to avoid locking the db. - # CLN might want to write while we read. - info = await get_ln_info_impl() - - # FIXME(#87): Once Core Lightnings accountability plugin is available - src = "/home/bitcoin/.lightning/bitcoin/lightningd.sqlite3" - dest = "/tmp/lightningd.sqlite3" - shutil.copyfile(src, dest) - - conn = sqlite3.connect(dest, uri=True) - cur = conn.execute("select * from outputs") - res = cur.fetchall() - conn.close() - - txs = [] - for o in res: - prev_out_tx = o[0].hex() - amount = o[2] - conf_block = o[9] - spent_block = o[10] - conf_time = (await _get_block_time(conf_block))[0] - tx_hash = f"prev_out_tx {prev_out_tx}" - - confs = info.block_height - conf_block - if confs < 0: - confs = 0 - logging.error( - f"Got negative confirmation count of for {tx_hash}\nCalc:{info.block_height} - {conf_block} = {confs}" - ) - - txs.append( - OnChainTransaction( - tx_hash=tx_hash, - amount=amount, - num_confirmations=confs, - block_height=conf_block, - time_stamp=conf_time, - total_fees=0, - ) - ) - - if spent_block is not None: - spent_time = (await _get_block_time(spent_block))[0] txs.append( OnChainTransaction( tx_hash=tx_hash, - amount=-amount, + amount=amount, num_confirmations=confs, - block_height=spent_block, - time_stamp=spent_time, + block_height=conf_block, + time_stamp=conf_time, total_fees=0, - ), - ) - - return txs - - -async def list_payments_impl( - include_incomplete: bool, index_offset: int, max_payments: int, reversed: bool -): - logging.debug( - f"CLN_GRPC: list_payments_impl(include_incomplete={include_incomplete}, index_offset{index_offset}, max_payments={max_payments}, reversed={reversed})" - ) - - req = ln.ListpaysRequest() - res = await _cln_stub.ListPays(req) - - pays = [] - for p in res.pays: - if p.status == 2: - # always include completed payments - pays.append(Payment.from_cln_grpc(p)) - continue - - if include_incomplete: - pays.append(Payment.from_cln_grpc(p)) - - if reversed: - pays.reverse() - - if max_payments == 0 or max_payments is None: - return pays - - return pays[index_offset : index_offset + max_payments] - - -async def add_invoice_impl( - value_msat: int, memo: str = "", expiry: int = 3600, is_keysend: bool = False -) -> Invoice: - logging.debug( - f"CLN_GRPC: add_invoice_impl(value_msat={value_msat}, memo={memo}, expiry={expiry}, is_keysend={is_keysend})" - ) - - if value_msat < 0: - raise ValueError("value_msat cannot be negative") - - msat = None - if value_msat == 0: - msat = lnp.AmountOrAny(any=True) - elif value_msat > 0: - msat = lnp.AmountOrAny(amount=lnp.Amount(msat=value_msat)) - - id = next_push_id() - req = ln.InvoiceRequest( - msatoshi=msat, - description=memo, - label=id, - expiry=expiry, - ) - - res = await _cln_stub.Invoice(req) - - return Invoice( - payment_request=res.bolt11, - memo=memo, - value_msat=value_msat, - expiry_date=res.expires_at, - add_index=id, - state=InvoiceState.OPEN, - ) - - -async def decode_pay_request_impl(pay_req: str) -> PaymentRequest: - logging.debug(f"CLN_GRPC: decode_pay_request_impl(pay_req={pay_req})") - - res = await _make_local_call(f"decodepay bolt11={pay_req}") - - if not res: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="Unknown CLN error decoding pay request", - ) - - if len(res) == 0: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="No response from CLN decoding pay request", - ) - - decoded = res[0].decode() - - if "Invalid bolt11: Bad bech32 string" in decoded: - raise HTTPException( - status.HTTP_400_BAD_REQUEST, detail="Invalid bolt11: Bad bech32 string" - ) - - return PaymentRequest.from_cln_json(json.loads(decoded)) - - -async def get_fee_revenue_impl() -> FeeRevenue: - logging.debug(f"CLN_GRPC: get_fee_revenue_impl()") - - # status 1 == "settled" - req = ln.ListforwardsRequest(status=1) - res = await _cln_stub.ListForwards(req) - - day = week = month = year = total = 0 - - now = time.time() - t_day = now - 86400.0 # 1 day - t_week = now - 604800.0 # 1 week - t_month = now - 2592000.0 # 1 month - t_year = now - 31536000.0 # 1 year - - # TODO: performance: cache this in redis - for f in res.forwards: - received_time = f.received_time - fee = f.fee_msat.msat - total += fee - - if received_time > t_day: - day += fee - week += fee - month += fee - year += fee - elif received_time > t_week: - week += fee - month += fee - year += fee - elif received_time > t_month: - month += fee - year += fee - elif received_time > t_year: - year += fee - - return FeeRevenue(day=day, week=week, month=month, year=year, total=total) - - -async def new_address_impl(input: NewAddressInput) -> str: - logging.debug(f"CLN_GRPC: new_address_impl(input={input})") - - if input.type == OnchainAddressType.P2WKH: - req = ln.NewaddrRequest(addresstype=2) - res = await _cln_stub.NewAddr(req) - return res.bech32 - - req = ln.NewaddrRequest(addresstype=1) - res = await _cln_stub.NewAddr(req) - return res.p2sh_segwit - - -async def send_coins_impl(input: SendCoinsInput) -> SendCoinsResponse: - logging.debug(f"CLN_GRPC: send_coins_impl(input={input})") - - fee_rate: lnp.Feerate = None - if input.sat_per_vbyte != None and input.sat_per_vbyte > 0: - fee_rate = lnp.Feerate(perkw=input.sat_per_vbyte) - elif input.target_conf != None and input.target_conf == 1: - fee_rate = lnp.Feerate(urgent=True) - elif input.target_conf != None and input.target_conf >= 2: - fee_rate = lnp.Feerate(normal=True) - elif input.target_conf != None and input.target_conf >= 10: - fee_rate = lnp.Feerate(slow=True) - - try: - funds = await _cln_stub.ListFunds(ln.ListfundsRequest()) - if len(funds.outputs) == 0: - raise HTTPException( - status.HTTP_412_PRECONDITION_FAILED, - detail=f"Could not afford {input.amount}sat. No UTXOs available at all", - ) - - utxos = [] - max_amt = 0 - for o in funds.outputs: - utxos.append(lnp.Outpoint(txid=o.txid, outnum=o.output)) - max_amt += o.amount_msat.msat - - if max_amt <= input.amount: - raise HTTPException( - status.HTTP_412_PRECONDITION_FAILED, - detail=f"Could not afford {input.amount}sat. Not enough funds available", - ) - - req = ln.WithdrawRequest( - destination=input.address, - satoshi=lnp.AmountOrAll(amount=lnp.Amount(msat=input.amount), all=False), - minconf=input.min_confs, - feerate=fee_rate, - utxos=utxos, - ) - res = await _cln_stub.Withdraw(req) - return SendCoinsResponse.from_cln_grpc(res, input) - except grpc.aio._call.AioRpcError as error: - details = error.details() - if details and details.find("Could not parse destination address") > -1: - raise HTTPException( - status.HTTP_400_BAD_REQUEST, - detail="Could not parse destination address, destination should be a valid address.", - ) - elif ( - details - and details.find("UTXO") > -1 - and details.find("already reserved") > -1 - ): - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="Server tried to use a reserved UTXO. Please submit an issue to the BlitzAPI repository.", - ) - elif details and details.find("insufficient funds available") > -1: - raise HTTPException(status.HTTP_412_PRECONDITION_FAILED, detail=details) - else: - raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=details) - - -async def send_payment_impl( - pay_req: str, - timeout_seconds: int, - fee_limit_msat: int, - amount_msat: Optional[int] = None, -) -> Payment: - logging.debug( - f"CLN_GRPC: send_payment_impl(pay_req={pay_req}, timeout_seconds={timeout_seconds}, fee_limit_msat={fee_limit_msat}, amount_msat={amount_msat})" - ) - - amt = lnp.Amount(msat=amount_msat) if amount_msat != None else None - fee_limit = lnp.Amount(msat=fee_limit_msat) - req = ln.PayRequest( - bolt11=pay_req, - msatoshi=amt, - maxfee=fee_limit, - retry_for=timeout_seconds, - ) - - try: - res = await _cln_stub.Pay(req) - except grpc.aio._call.AioRpcError as error: - details = error.details() - - if "Ran out of routes to try after" in details: - attempts = details.split("Ran out of routes to try after ")[1] - attempts = attempts.split(" attempts")[0] - raise HTTPException( - status.HTTP_422_UNPROCESSABLE_ENTITY, - detail=f"Ran out of routes to try after {attempts} attempts.", - ) - - if "msatoshi parameter required" in details: - raise HTTPException( - status.HTTP_400_BAD_REQUEST, - detail="amount must be specified when paying a zero amount invoice", - ) - - if "msatoshi parameter unnecessary" in details: - raise HTTPException( - status.HTTP_400_BAD_REQUEST, - detail="amount must not be specified when paying a non-zero amount invoice", - ) - - raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=details) - - return Payment.from_cln_grpc(res) - - -async def get_ln_info_impl() -> LnInfo: - logging.debug(f"CLN_GRPC: get_ln_info_impl()") - - req = ln.GetinfoRequest() - res = await _cln_stub.Getinfo(req) - return LnInfo.from_cln_grpc(get_implementation_name(), res) - - -async def unlock_wallet_impl(password: str) -> bool: - logging.debug(f"CLN_GRPC: unlock_wallet_impl(password=wedontlogpasswords)") - - # Core Lightning doesn't lock wallets, - # so we don't need to do anything here - return True - - -async def listen_invoices() -> AsyncGenerator[Invoice, None]: - logging.debug(f"CLN_GRPC: listen_invoices()") - - lastpay_index = 0 - invoices = await list_invoices_impl( - pending_only=False, - index_offset=0, - num_max_invoices=9999999999999, - reversed=False, - ) - - for i in invoices: # type Invoice - if i.state == InvoiceState.SETTLED and i.settle_index > lastpay_index: - lastpay_index = i.settle_index - - while True: - req = ln.WaitanyinvoiceRequest(lastpay_index=lastpay_index) - i = await _cln_stub.WaitAnyInvoice(req) - i = Invoice.from_cln_grpc(i) - lastpay_index = i.settle_index - yield i - - -async def listen_forward_events() -> ForwardSuccessEvent: - logging.debug(f"CLN_GRPC: listen_forward_events()") - - # CLN has no subscription to forwarded events. - # We must poll instead. - - interval = config("gather_ln_info_interval", default=2, cast=float) - - # make sure we know how many forwards we have - # we need to calculate the difference between each iteration - # status=1 == "settled" - req = ln.ListforwardsRequest(status=1) - res = await _cln_stub.ListForwards(req) - num_fwd_last_poll = len(res.forwards) - while True: - res = await _cln_stub.ListForwards(req) - if len(res.forwards) > num_fwd_last_poll: - fwds = res.forwards[num_fwd_last_poll:] - for fwd in fwds: - yield ForwardSuccessEvent.from_cln_grpc(fwd) - - num_fwd_last_poll = len(res.forwards) - await asyncio.sleep(interval - 0.1) - - -async def connect_peer_impl(node_URI: str) -> bool: - logging.debug(f"CLN_GRPC: connect_peer_impl(node_URI={node_URI})") - - try: - id = node_URI.split("@")[0] - stdout, stderr = await _make_local_call(f"connect id={node_URI}") - if stdout: - if id in stdout.decode(): - return True - if "Connection timed out" in stdout.decode(): - raise HTTPException( - status.HTTP_504_GATEWAY_TIMEOUT, - detail="Connection establishment: Connection timed out.", ) - if "Connection refused" in stdout.decode(): - raise HTTPException( - status.HTTP_504_GATEWAY_TIMEOUT, - detail="Connection establishment: Connection refused.", + ) + + if spent_block is not None: + spent_time = (await self._get_block_time(spent_block))[0] + txs.append( + OnChainTransaction( + tx_hash=tx_hash, + amount=-amount, + num_confirmations=confs, + block_height=spent_block, + time_stamp=spent_time, + total_fees=0, + ), ) - if stderr: - logging.error(f"CLN_GRPC: {stderr.decode()}") - return False + return txs - except grpc.aio._call.AioRpcError as error: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + async def list_payments( + self, + include_incomplete: bool, + index_offset: int, + max_payments: int, + reversed: bool, + ): + logging.debug( + f"CLN_GRPC: list_payments(include_incomplete={include_incomplete}, index_offset{index_offset}, max_payments={max_payments}, reversed={reversed})" ) + req = ln.ListpaysRequest() + res = await self._cln_stub.ListPays(req) -async def peer_resolve_alias(node_pub: str) -> str: - logging.debug(f"CLN_GRPC: peer_resolve_alias(node_pub={node_pub})") + pays = [] + for p in res.pays: + if p.status == 2: + # always include completed payments + pays.append(Payment.from_cln_grpc(p)) + continue - try: - request = ln.ListnodesRequest(id=node_pub) - response = await _cln_stub.ListNodes(request) + if include_incomplete: + pays.append(Payment.from_cln_grpc(p)) - if len(response.nodes) == 0: - return "" + if reversed: + pays.reverse() - return str(response.nodes[0].alias) + if max_payments == 0 or max_payments is None: + return pays - except grpc.aio._call.AioRpcError as error: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + return pays[index_offset : index_offset + max_payments] + + async def add_invoice( + self, + value_msat: int, + memo: str = "", + expiry: int = 3600, + is_keysend: bool = False, + ) -> Invoice: + logging.debug( + f"CLN_GRPC: add_invoice(value_msat={value_msat}, memo={memo}, expiry={expiry}, is_keysend={is_keysend})" ) + if value_msat < 0: + raise ValueError("value_msat cannot be negative") -async def channel_open_impl( - local_funding_amount: int, node_URI: str, target_confs: int -) -> str: - logging.debug( - f"CLN_GRPC: channel_open_impl(local_funding_amount={local_funding_amount}, node_URI={node_URI}, target_confs={target_confs})" - ) + msat = None + if value_msat == 0: + msat = lnp.AmountOrAny(any=True) + elif value_msat > 0: + msat = lnp.AmountOrAny(amount=lnp.Amount(msat=value_msat)) - fee_rate = None - if target_confs == 1: - fee_rate = "urgent" - elif target_confs >= 2 and target_confs <= 9: - fee_rate = "normal" - elif target_confs >= 10: - fee_rate = "slow" + id = next_push_id() + req = ln.InvoiceRequest( + msatoshi=msat, + description=memo, + label=id, + expiry=expiry, + ) + + try: + res = await self._cln_stub.Invoice(req) + except Exception as e: + print(e) + + return Invoice( + payment_request=res.bolt11, + memo=memo, + value_msat=value_msat, + expiry_date=res.expires_at, + add_index=id, + state=InvoiceState.OPEN, + ) + + async def decode_pay_request(self, pay_req: str) -> PaymentRequest: + logging.debug(f"CLN_GRPC: decode_pay_request(pay_req={pay_req})") + + res = await _make_local_call(f"decodepay bolt11={pay_req}") - try: - res = await connect_peer_impl(node_URI) if not res: raise HTTPException( - status.HTTP_408_REQUEST_TIMEOUT, - detail="Unknown error while trying to connect to peer", + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Unknown CLN error decoding pay request", ) - cmd = f"fundchannel id={node_URI} amount={local_funding_amount} feerate={fee_rate}" - stdout, stderr = await _make_local_call(cmd) - if stdout: - o = stdout.decode() - j = json.loads(o) - if "txid" in o and "channel_id" in o: - return j["txid"] - if "Unknown peer" in o: + if len(res) == 0: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="No response from CLN decoding pay request", + ) + + decoded = res[0].decode() + + if "Invalid bolt11: Bad bech32 string" in decoded: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, detail="Invalid bolt11: Bad bech32 string" + ) + + return PaymentRequest.from_cln_json(json.loads(decoded)) + + async def get_fee_revenue(self) -> FeeRevenue: + logging.debug(f"CLN_GRPC: get_fee_revenue()") + + # status 1 == "settled" + req = ln.ListforwardsRequest(status=1) + res = await self._cln_stub.ListForwards(req) + + day = week = month = year = total = 0 + + now = time.time() + t_day = now - 86400.0 # 1 day + t_week = now - 604800.0 # 1 week + t_month = now - 2592000.0 # 1 month + t_year = now - 31536000.0 # 1 year + + # TODO: performance: cache this in redis + for f in res.forwards: + received_time = f.received_time + fee = f.fee_msat.msat + total += fee + + if received_time > t_day: + day += fee + week += fee + month += fee + year += fee + elif received_time > t_week: + week += fee + month += fee + year += fee + elif received_time > t_month: + month += fee + year += fee + elif received_time > t_year: + year += fee + + return FeeRevenue(day=day, week=week, month=month, year=year, total=total) + + async def new_address(self, input: NewAddressInput) -> str: + logging.debug(f"CLN_GRPC: new_address(input={input})") + + if input.type == OnchainAddressType.P2WKH: + req = ln.NewaddrRequest(addresstype=2) + res = await self._cln_stub.NewAddr(req) + return res.bech32 + + req = ln.NewaddrRequest(addresstype=1) + res = await self._cln_stub.NewAddr(req) + return res.p2sh_segwit + + async def send_coins(self, input: SendCoinsInput) -> SendCoinsResponse: + logging.debug(f"CLN_GRPC: send_coins(input={input})") + + fee_rate: lnp.Feerate = None + if input.sat_per_vbyte != None and input.sat_per_vbyte > 0: + fee_rate = lnp.Feerate(perkw=input.sat_per_vbyte) + elif input.target_conf != None and input.target_conf == 1: + fee_rate = lnp.Feerate(urgent=True) + elif input.target_conf != None and input.target_conf >= 2: + fee_rate = lnp.Feerate(normal=True) + elif input.target_conf != None and input.target_conf >= 10: + fee_rate = lnp.Feerate(slow=True) + + try: + funds = await self._cln_stub.ListFunds(ln.ListfundsRequest()) + if len(funds.outputs) == 0: raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="We where able to connect to the peer but CLN can't find it when opening a channel.", + status.HTTP_412_PRECONDITION_FAILED, + detail=f"Could not afford {input.amount}sat. No UTXOs available at all", ) - if "Owning subdaemon openingd died" in o: - # https://github.com/ElementsProject/lightning/issues/2798#issuecomment-511205719 + + utxos = [] + max_amt = 0 + for o in funds.outputs: + utxos.append(lnp.Outpoint(txid=o.txid, outnum=o.output)) + max_amt += o.amount_msat.msat + + if max_amt <= input.amount: + raise HTTPException( + status.HTTP_412_PRECONDITION_FAILED, + detail=f"Could not afford {input.amount}sat. Not enough funds available", + ) + + req = ln.WithdrawRequest( + destination=input.address, + satoshi=lnp.AmountOrAll( + amount=lnp.Amount(msat=input.amount), all=False + ), + minconf=input.min_confs, + feerate=fee_rate, + utxos=utxos, + ) + res = await self._cln_stub.Withdraw(req) + return SendCoinsResponse.from_cln_grpc(res, input) + except grpc.aio._call.AioRpcError as error: + details = error.details() + if details and details.find("Could not parse destination address") > -1: raise HTTPException( status.HTTP_400_BAD_REQUEST, - detail="Likely the peer didn't like our channel opening proposal and disconnected from us.", + detail="Could not parse destination address, destination should be a valid address.", + ) + elif ( + details + and details.find("UTXO") > -1 + and details.find("already reserved") > -1 + ): + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Server tried to use a reserved UTXO. Please submit an issue to the BlitzAPI repository.", + ) + elif details and details.find("insufficient funds available") > -1: + raise HTTPException(status.HTTP_412_PRECONDITION_FAILED, detail=details) + else: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=details ) - if "Could not afford " in o: - raise HTTPException(status.HTTP_400_BAD_REQUEST, detail=j["message"]) - if "Number of pending channels exceed maximum" in o: - raise HTTPException(status.HTTP_400_BAD_REQUEST, detail=j["message"]) + async def send_payment( + self, + pay_req: str, + timeout_seconds: int, + fee_limit_msat: int, + amount_msat: Optional[int] = None, + ) -> Payment: + logging.debug( + f"CLN_GRPC: send_payment(pay_req={pay_req}, timeout_seconds={timeout_seconds}, fee_limit_msat={fee_limit_msat}, amount_msat={amount_msat})" + ) + + amt = lnp.Amount(msat=amount_msat) if amount_msat != None else None + fee_limit = lnp.Amount(msat=fee_limit_msat) + req = ln.PayRequest( + bolt11=pay_req, + msatoshi=amt, + maxfee=fee_limit, + retry_for=timeout_seconds, + ) + + try: + res = await self._cln_stub.Pay(req) + except grpc.aio._call.AioRpcError as error: + details = error.details() + + if "Ran out of routes to try after" in details: + attempts = details.split("Ran out of routes to try after ")[1] + attempts = attempts.split(" attempts")[0] + raise HTTPException( + status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"Ran out of routes to try after {attempts} attempts.", + ) + + if "msatoshi parameter required" in details: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + detail="amount must be specified when paying a zero amount invoice", + ) + + if "msatoshi parameter unnecessary" in details: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + detail="amount must not be specified when paying a non-zero amount invoice", + ) + + raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=details) + + return Payment.from_cln_grpc(res) + + async def get_ln_info(self) -> LnInfo: + logging.debug(f"CLN_GRPC: get_ln_info()") + + req = ln.GetinfoRequest() + res = await self._cln_stub.Getinfo(req) + return LnInfo.from_cln_grpc(self.get_implementation_name(), res) + + async def unlock_wallet(self, password: str) -> bool: + logging.debug(f"CLN_GRPC: unlock_wallet(password=wedontlogpasswords)") + + # Core Lightning doesn't lock wallets, + # so we don't need to do anything here + return True + + async def listen_invoices(self) -> AsyncGenerator[Invoice, None]: + logging.debug(f"CLN_GRPC: listen_invoices()") + + lastpay_index = 0 + invoices = await self.list_invoices( + pending_only=False, + index_offset=0, + num_max_invoices=9999999999999, + reversed=False, + ) + + for i in invoices: # type Invoice + if i.state == InvoiceState.SETTLED and i.settle_index > lastpay_index: + lastpay_index = i.settle_index + + while True: + req = ln.WaitanyinvoiceRequest(lastpay_index=lastpay_index) + i = await self._cln_stub.WaitAnyInvoice(req) + i = Invoice.from_cln_grpc(i) + lastpay_index = i.settle_index + yield i + + async def listen_forward_events(self) -> ForwardSuccessEvent: + logging.debug(f"CLN_GRPC: listen_forward_events()") + + # CLN has no subscription to forwarded events. + # We must poll instead. + + interval = config("gather_ln_info_interval", default=2, cast=float) + + # make sure we know how many forwards we have + # we need to calculate the difference between each iteration + # status=1 == "settled" + req = ln.ListforwardsRequest(status=1) + res = await self._cln_stub.ListForwards(req) + num_fwd_last_poll = len(res.forwards) + while True: + res = await self._cln_stub.ListForwards(req) + if len(res.forwards) > num_fwd_last_poll: + fwds = res.forwards[num_fwd_last_poll:] + for fwd in fwds: + yield ForwardSuccessEvent.from_cln_grpc(fwd) + + num_fwd_last_poll = len(res.forwards) + await asyncio.sleep(interval - 0.1) + + async def connect_peer(self, node_URI: str) -> bool: + logging.debug(f"CLN_GRPC: connect_peer(node_URI={node_URI})") + + try: + id = node_URI.split("@")[0] + stdout, stderr = await _make_local_call(f"connect id={node_URI}") + if stdout: + if id in stdout.decode(): + return True + if "Connection timed out" in stdout.decode(): + raise HTTPException( + status.HTTP_504_GATEWAY_TIMEOUT, + detail="Connection establishment: Connection timed out.", + ) + if "Connection refused" in stdout.decode(): + raise HTTPException( + status.HTTP_504_GATEWAY_TIMEOUT, + detail="Connection establishment: Connection refused.", + ) + if stderr: + logging.error(f"CLN_GRPC: {stderr.decode()}") + + return False + + except grpc.aio._call.AioRpcError as error: raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=j["message"] - ) - if stderr: - logging.error(f"CLN_GRPC: {stderr.decode()}") - - return False - except grpc.aio._call.AioRpcError as error: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def channel_list_impl() -> List[Channel]: - logging.debug(f"CLN_GRPC: channel_list_impl()") - - try: - res = await _cln_stub.ListFunds(ln.ListfundsRequest()) - peer_ids = [c.peer_id for c in res.channels] - peer_res = await asyncio.gather(*[peer_resolve_alias(p) for p in peer_ids]) - channels = [Channel.from_cln_grpc(c, p) for c, p in zip(res.channels, peer_res)] - return channels - except grpc.aio._call.AioRpcError as error: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def channel_close_impl(channel_id: int, force_close: bool) -> str: - logging.debug( - f"CLN_GRPC: channel_close_impl(channel_id={channel_id}, force_close={force_close})" - ) - - try: - # on CLN we wait for 2 minutes to negotiate a channel close - # if peer doesn't respond we force close - wait_time_before_unilateral_close = 120 if force_close else 0 - req = ln.CloseRequest( - id=channel_id, - unilateraltimeout=wait_time_before_unilateral_close, - feerange=[lnp.Feerate(slow=True), lnp.Feerate(urgent=True)], - ) - res = await _cln_stub.Close(req) - - # “mutual”, “unilateral”, “unopened” - t = res.item_type - if t == 0 or t == 1: # mutual, unilateral - return res.txid.hex() - elif t == 2: # unopened - raise HTTPException( - status.HTTP_400_BAD_REQUEST, detail="Channel is not open yet." + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() ) - raise HTTPException( - status.HTTP_400_BAD_REQUEST, - detail=f"CLN returned unknown close type: {t}", + async def peer_resolve_alias(self, node_pub: str) -> str: + logging.debug(f"CLN_GRPC: peer_resolve_alias(node_pub={node_pub})") + + try: + request = ln.ListnodesRequest(id=node_pub) + response = await self._cln_stub.ListNodes(request) + + if len(response.nodes) == 0: + return "" + + return str(response.nodes[0].alias) + + except grpc.aio._call.AioRpcError as error: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def channel_open( + self, local_funding_amount: int, node_URI: str, target_confs: int + ) -> str: + logging.debug( + f"CLN_GRPC: channel_open(local_funding_amount={local_funding_amount}, node_URI={node_URI}, target_confs={target_confs})" ) - except grpc.aio._call.AioRpcError as error: - if "Channel is in state AWAITING_UNILATERAL" in error.details(): + + fee_rate = None + if target_confs == 1: + fee_rate = "urgent" + elif target_confs >= 2 and target_confs <= 9: + fee_rate = "normal" + elif target_confs >= 10: + fee_rate = "slow" + + try: + res = await self.connect_peer(node_URI) + if not res: + raise HTTPException( + status.HTTP_408_REQUEST_TIMEOUT, + detail="Unknown error while trying to connect to peer", + ) + + cmd = f"fundchannel id={node_URI} amount={local_funding_amount} feerate={fee_rate}" + stdout, stderr = await _make_local_call(cmd) + if stdout: + o = stdout.decode() + j = json.loads(o) + if "txid" in o and "channel_id" in o: + return j["txid"] + if "Unknown peer" in o: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="We where able to connect to the peer but CLN can't find it when opening a channel.", + ) + if "Owning subdaemon openingd died" in o: + # https://github.com/ElementsProject/lightning/issues/2798#issuecomment-511205719 + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + detail="Likely the peer didn't like our channel opening proposal and disconnected from us.", + ) + if "Could not afford " in o: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, detail=j["message"] + ) + if "Number of pending channels exceed maximum" in o: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, detail=j["message"] + ) + + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=j["message"] + ) + if stderr: + logging.error(f"CLN_GRPC: {stderr.decode()}") + + return False + except grpc.aio._call.AioRpcError as error: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def channel_list(self) -> List[Channel]: + logging.debug(f"CLN_GRPC: channel_list()") + + try: + res = await self._cln_stub.ListFunds(ln.ListfundsRequest()) + peer_ids = [c.peer_id for c in res.channels] + peer_res = await asyncio.gather( + *[self.peer_resolve_alias(p) for p in peer_ids] + ) + channels = [ + Channel.from_cln_grpc(c, p) for c, p in zip(res.channels, peer_res) + ] + return channels + except grpc.aio._call.AioRpcError as error: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def channel_close(self, channel_id: int, force_close: bool) -> str: + logging.debug( + f"CLN_GRPC: channel_close(channel_id={channel_id}, force_close={force_close})" + ) + + try: + # on CLN we wait for 2 minutes to negotiate a channel close + # if peer doesn't respond we force close + wait_time_before_unilateral_close = 120 if force_close else 0 + req = ln.CloseRequest( + id=channel_id, + unilateraltimeout=wait_time_before_unilateral_close, + feerange=[lnp.Feerate(slow=True), lnp.Feerate(urgent=True)], + ) + res = await self._cln_stub.Close(req) + + # “mutual”, “unilateral”, “unopened” + t = res.item_type + if t == 0 or t == 1: # mutual, unilateral + return res.txid.hex() + elif t == 2: # unopened + raise HTTPException( + status.HTTP_400_BAD_REQUEST, detail="Channel is not open yet." + ) + raise HTTPException( status.HTTP_400_BAD_REQUEST, - detail="Channel is awaiting an unilateral close.", + detail=f"CLN returned unknown close type: {t}", ) + except grpc.aio._call.AioRpcError as error: + if "Channel is in state AWAITING_UNILATERAL" in error.details(): + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + detail="Channel is awaiting an unilateral close.", + ) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) diff --git a/app/repositories/ln_impl/ln_base.py b/app/repositories/ln_impl/ln_base.py new file mode 100644 index 0000000..ee46116 --- /dev/null +++ b/app/repositories/ln_impl/ln_base.py @@ -0,0 +1,133 @@ +from abc import abstractmethod +from typing import AsyncGenerator, List, Optional + +from app.models.lightning import ( + Channel, + FeeRevenue, + ForwardSuccessEvent, + GenericTx, + InitLnRepoUpdate, + Invoice, + LnInfo, + NewAddressInput, + OnChainTransaction, + Payment, + PaymentRequest, + SendCoinsInput, + SendCoinsResponse, + WalletBalance, +) + + +class LightningNodeBase: + @abstractmethod + def get_implementation_name(self) -> str: + raise NotImplementedError() + + @abstractmethod + async def initialize(self) -> AsyncGenerator[InitLnRepoUpdate, None]: + raise NotImplementedError() + + @abstractmethod + async def get_wallet_balance(self) -> WalletBalance: + raise NotImplementedError() + + @abstractmethod + async def list_all_tx( + self, successful_only: bool, index_offset: int, max_tx: int, reversed: bool + ) -> List[GenericTx]: + raise NotImplementedError() + + @abstractmethod + async def list_invoices( + self, + pending_only: bool, + index_offset: int, + num_max_invoices: int, + reversed: bool, + ): + raise NotImplementedError() + + @abstractmethod + async def list_on_chain_tx(self) -> List[OnChainTransaction]: + raise NotImplementedError() + + @abstractmethod + async def list_payments( + self, + include_incomplete: bool, + index_offset: int, + max_payments: int, + reversed: bool, + ): + raise NotImplementedError() + + @abstractmethod + async def add_invoice( + self, + value_msat: int, + memo: str = "", + expiry: int = 3600, + is_keysend: bool = False, + ) -> Invoice: + raise NotImplementedError() + + @abstractmethod + async def decode_pay_request(self, pay_req: str) -> PaymentRequest: + raise NotImplementedError() + + @abstractmethod + async def get_fee_revenue(self) -> FeeRevenue: + raise NotImplementedError() + + @abstractmethod + async def new_address(self, input: NewAddressInput) -> str: + raise NotImplementedError() + + @abstractmethod + async def send_coins(self, input: SendCoinsInput) -> SendCoinsResponse: + raise NotImplementedError() + + @abstractmethod + async def send_payment( + self, + pay_req: str, + timeout_seconds: int, + fee_limit_msat: int, + amount_msat: Optional[int] = None, + ) -> Payment: + raise NotImplementedError() + + @abstractmethod + async def get_ln_info(self) -> LnInfo: + raise NotImplementedError() + + @abstractmethod + async def unlock_wallet(self, password: str) -> bool: + raise NotImplementedError() + + @abstractmethod + async def listen_invoices(self) -> AsyncGenerator[Invoice, None]: + raise NotImplementedError() + + @abstractmethod + async def listen_forward_events(self) -> ForwardSuccessEvent: + raise NotImplementedError() + + @abstractmethod + async def channel_open( + self, local_funding_amount: int, node_URI: str, target_confs: int + ) -> str: + raise NotImplementedError() + + @abstractmethod + async def peer_resolve_alias(self, node_pub: str) -> str: + raise NotImplementedError() + + @abstractmethod + async def channel_list(self) -> List[Channel]: + raise NotImplementedError() + + @abstractmethod + async def channel_close(self, channel_id: int, force_close: bool) -> str: + raise NotImplementedError() diff --git a/app/repositories/ln_impl/lnd_grpc.py b/app/repositories/ln_impl/lnd_grpc.py index 1760fa3..03aba20 100644 --- a/app/repositories/ln_impl/lnd_grpc.py +++ b/app/repositories/ln_impl/lnd_grpc.py @@ -34,687 +34,7 @@ from app.models.lightning import ( SendCoinsResponse, WalletBalance, ) - -_lnd_connect_error_debug_msg = """ -LND_GRPC: Unable to connect to LND. Possible reasons: -* Node is not reachable (ports, network down, ...) -* Macaroon is not correct -* IP is not included in LND tls certificate - Add tlsextraip=192.168.1.xxx to lnd.conf and restart LND. - This will recreate the TLS certificate. The .env must be adapted accordingly. -* TLS certificate is wrong. (settings changed, ...) - -To Debug gRPC problems uncomment the following line in app.repositories.ln_impl.lnd_grpc.py -# os.environ["GRPC_VERBOSITY"] = "DEBUG" -This will show more debug information. -""" - -_initialized = False - -# Due to updated ECDSA generated tls.cert we need to let gprc know that -# we need to use that cipher suite otherwise there will be a handshake -# error when we communicate with the lnd rpc server. -os.environ["GRPC_SSL_CIPHER_SUITES"] = "HIGH+ECDSA" - -# Uncomment to see full gRPC logs -# os.environ["GRPC_TRACE"] = "all" -# os.environ["GRPC_VERBOSITY"] = "DEBUG" - - -def _metadata_callback(context, callback): - # for more info see grpc docs - callback([("macaroon", _lnd_macaroon)], None) - - -_lnd_macaroon = config_get_hex_str(dconfig("lnd_macaroon"), name="lnd_macaroon") -_lnd_cert = bytes.fromhex(config_get_hex_str(dconfig("lnd_cert"), name="lnd_cert")) -_lnd_grpc_ip = dconfig("lnd_grpc_ip") -_lnd_grpc_port = dconfig("lnd_grpc_port") -_lnd_grpc_url = _lnd_grpc_ip + ":" + _lnd_grpc_port - -_auth_creds = grpc.metadata_call_credentials(_metadata_callback) -_ssl_creds = grpc.ssl_channel_credentials(_lnd_cert) -_combined_creds = grpc.composite_channel_credentials(_ssl_creds, _auth_creds) -_channel = None -_lnd_stub = None -_router_stub = None -_wallet_unlocker = None - - -def _create_stubs() -> None: - global _channel - global _lnd_stub - global _router_stub - global _wallet_unlocker - - if _channel is not None: - logging.warning("LND_GRPC: gRPC channel already created.") - return - - _channel = grpc.aio.secure_channel(_lnd_grpc_url, _combined_creds) - _lnd_stub = lnrpc.LightningStub(_channel) - _router_stub = routerrpc.RouterStub(_channel) - _wallet_unlocker = unlockerrpc.WalletUnlockerStub(_channel) - - logging.debug("LND_GRPC: Created LND gRPC stubs") - - -def get_implementation_name() -> str: - return "LND_GRPC" - - -init_queue = asyncio.Queue() - - -async def _check_lnd_status( - sleep_time: float = 2, -) -> AsyncGenerator[InitLnRepoUpdate, None]: - logging.debug("LND_GRPC: _check_lnd_status() start") - - _lnd_connect_error_debug_msg_sent = False - - # Create a temporary channel which will be destroyed at each iteration - # Reason is that gRPC seems to only try and connect every 5 seconds to - # the node if it is not running. To avoid the delay we create a new - # channel each iteration. - - global _channel - global _lnd_stub - - channel = None - lnd_stub = None - while True: - try: - if channel is None: - if _channel is not None: - channel = _channel - lnd_stub = _lnd_stub - else: - channel = grpc.aio.secure_channel(_lnd_grpc_url, _combined_creds) - lnd_stub = lnrpc.LightningStub(channel) - await lnd_stub.GetInfo(ln.GetInfoRequest()) - - if _channel is None: - _create_stubs() - - await init_queue.put(InitLnRepoUpdate(state=LnInitState.DONE)) - break - except grpc.aio._call.AioRpcError as error: - details = error.details() - logging.debug(f"LND_GRPC: Waiting for LND daemon... Details {details}") - - if "failed to connect to all addresses" in details: - await init_queue.put( - InitLnRepoUpdate( - state=LnInitState.OFFLINE, - msg="Unable to connect to LND daemon, waiting...", - ) - ) - - if not _lnd_connect_error_debug_msg_sent: - logging.debug(_lnd_connect_error_debug_msg) - _lnd_connect_error_debug_msg_sent = True - - await channel.close() - channel = None - elif "waiting to start, RPC services not available" in details: - await init_queue.put( - InitLnRepoUpdate( - state=LnInitState.BOOTSTRAPPING, - msg="Connected but waiting to start, RPC services not available", - ) - ) - await channel.close() - channel = None - elif "wallet locked, unlock it to enable full RPC access" in details: - await init_queue.put( - InitLnRepoUpdate( - state=LnInitState.LOCKED, - msg="Wallet locked, unlock it to enable full RPC access", - ) - ) - if channel != _channel: - await channel.close() - channel = None - elif ( - "the RPC server is in the process of starting up, but not yet ready to accept calls" - in details - ): - # message from LND AFTER unlocking the wallet - await init_queue.put( - InitLnRepoUpdate( - state=LnInitState.BOOTSTRAPPING_AFTER_UNLOCK, - msg="The RPC server is in the process of starting up, but not yet ready to accept calls", - ) - ) - else: - logging.error(f"LND_GRPC: Unknown error: {details}") - raise - - logging.debug( - f"LND_GRPC: _check_lnd_status() sleeping {sleep_time} seconds..." - ) - await asyncio.sleep(sleep_time) - - logging.debug("LND_GRPC: _check_lnd_status() done") - - -async def initialize_impl() -> AsyncGenerator[InitLnRepoUpdate, None]: - logging.debug("LND_GRPC: Unable to connect to LND daemon, waiting...") - global _initialized - if _initialized: - logging.warning( - "LND_GRPC: Connection already initialized. This function must not be called twice." - ) - yield InitLnRepoUpdate(state=LnInitState.DONE) - - logging.info("LND_GRPC: Unable to connect to LND daemon, waiting...") - - global _channel - global _lnd_stub - global _router_stub - global _wallet_unlocker - - loop = asyncio.get_event_loop() - task = loop.create_task(_check_lnd_status(sleep_time=2)) - - while not _initialized: - res = await init_queue.get() # type: InitLnRepoUpdate - - if res.state == LnInitState.BOOTSTRAPPING_AFTER_UNLOCK and _channel is None: - task.cancel() - - if _channel == None: - # if res == _API_WALLET_UNLOCK_EVENT the endpoint function will have - # created the channel for us. - _create_stubs() - - task = loop.create_task(_check_lnd_status(sleep_time=0.5)) - elif res.state == LnInitState.DONE: - _initialized = True - if not task.cancelled(): - task.cancel() - elif ( - res.state == LnInitState.OFFLINE - or res.state == LnInitState.LOCKED - or res.state == LnInitState.BOOTSTRAPPING_AFTER_UNLOCK - ): - pass # do nothing here - else: - logging.warning(f"LND_GRPC: Unhandled initialization event: {res.dict()}") - - yield res - - logging.info("LND_GRPC: Initialization complete.") - - -async def get_wallet_balance_impl() -> WalletBalance: - logging.debug("LND_GRPC: get_wallet_balance_impl() ") - - try: - w_req = ln.WalletBalanceRequest() - onchain = await _lnd_stub.WalletBalance(w_req) - - c_req = ln.ChannelBalanceRequest() - channel = await _lnd_stub.ChannelBalance(c_req) - - return WalletBalance.from_lnd_grpc(onchain, channel) - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -# Decoding the payment request take a long time, -# hence we build a simple cache here. -memo_cache = {} - - -async def list_all_tx_impl( - successful_only: bool, index_offset: int, max_tx: int, reversed: bool -) -> List[GenericTx]: - logging.debug( - f"LND_GRPC: list_all_tx_impl(successful_only={successful_only}, index_offset={index_offset}, max_tx={max_tx}, reversed={reversed})" - ) - - # TODO: find a better caching strategy - list_invoice_req = ln.ListInvoiceRequest( - pending_only=successful_only, - index_offset=0, - num_max_invoices=0, - reversed=reversed, - ) - - get_tx_req = ln.GetTransactionsRequest() - - list_payments_req = ln.ListPaymentsRequest( - include_incomplete=not successful_only, - index_offset=0, - max_payments=0, - reversed=reversed, - ) - - try: - res = await asyncio.gather( - *[ - _lnd_stub.ListInvoices(list_invoice_req), - _lnd_stub.GetTransactions(get_tx_req), - _lnd_stub.ListPayments(list_payments_req), - ] - ) - - tx = [] - for i in res[0].invoices: - tx.append(GenericTx.from_lnd_grpc_invoice(i)) - for t in res[1].transactions: - tx.append(GenericTx.from_lnd_grpc_onchain_tx(t)) - for p in res[2].payments: - comment = "" - if p.payment_request in memo_cache: - comment = memo_cache[p.payment_request] - else: - if p.payment_request is not None and p.payment_request != "": - pr = await decode_pay_request_impl(p.payment_request) - comment = pr.description - memo_cache[p.payment_request] = pr.description - tx.append(GenericTx.from_lnd_grpc_payment(p, comment)) - - def sortKey(e: GenericTx): - return e.time_stamp - - tx.sort(key=sortKey) - - if reversed: - tx.reverse() - - l = len(tx) - for i in range(l): - tx[i].index = i - - if max_tx == 0: - max_tx = l - - return tx[index_offset : index_offset + max_tx] - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def list_invoices_impl( - pending_only: bool, index_offset: int, num_max_invoices: int, reversed: bool -): - logging.debug("LND_GRPC: list_invoices_impl() ") - - try: - req = ln.ListInvoiceRequest( - pending_only=pending_only, - index_offset=index_offset, - num_max_invoices=num_max_invoices, - reversed=reversed, - ) - response = await _lnd_stub.ListInvoices(req) - return [Invoice.from_lnd_grpc(i) for i in response.invoices] - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def list_on_chain_tx_impl() -> List[OnChainTransaction]: - logging.debug("LND_GRPC: list_on_chain_tx_impl() ") - - try: - req = ln.GetTransactionsRequest() - response = await _lnd_stub.GetTransactions(req) - return [OnChainTransaction.from_lnd_grpc(t) for t in response.transactions] - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def list_payments_impl( - include_incomplete: bool, index_offset: int, max_payments: int, reversed: bool -): - logging.debug( - f"LND_GRPC: list_payments_impl(include_incomplete={include_incomplete}, index_offset{index_offset}, max_payments={max_payments}, reversed={reversed})" - ) - - try: - req = ln.ListPaymentsRequest( - include_incomplete=include_incomplete, - index_offset=index_offset, - max_payments=max_payments, - reversed=reversed, - ) - response = await _lnd_stub.ListPayments(req) - return [Payment.from_lnd_grpc(p) for p in response.payments] - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def add_invoice_impl( - value_msat: int, memo: str = "", expiry: int = 3600, is_keysend: bool = False -) -> Invoice: - logging.debug( - f"LND_GRPC: add_invoice_impl(value_msat={value_msat}, memo={memo}, expiry={expiry}, is_keysend={is_keysend})" - ) - - try: - i = ln.Invoice( - memo=memo, - value_msat=value_msat, - expiry=expiry, - is_keysend=is_keysend, - ) - - response = await _lnd_stub.AddInvoice(i) - - # Can't use Invoice.from_lnd_grpc() here because - # the response is not a standard invoice - invoice = Invoice( - memo=memo, - expiry=expiry, - r_hash=response.r_hash.hex(), - payment_request=response.payment_request, - add_index=response.add_index, - payment_addr=response.payment_addr.hex(), - state=InvoiceState.OPEN, - is_keysend=is_keysend, - ) - - return invoice - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def decode_pay_request_impl(pay_req: str) -> PaymentRequest: - logging.debug(f"LND_GRPC: decode_pay_request_impl(pay_req={pay_req})") - - try: - req = ln.PayReqString(pay_req=pay_req) - res = await _lnd_stub.DecodePayReq(req) - return PaymentRequest.from_lnd_grpc(res) - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - if error.details() != None and error.details().find("checksum failed.") > -1: - raise HTTPException( - status.HTTP_400_BAD_REQUEST, detail="Invalid payment request string" - ) - else: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def get_fee_revenue_impl() -> FeeRevenue: - logging.debug(f"LND_GRPC: get_fee_revenue_impl()") - - req = ln.FeeReportRequest() - res = await _lnd_stub.FeeReport(req) - return FeeRevenue.from_lnd_grpc(res) - - -async def new_address_impl(input: NewAddressInput) -> str: - logging.debug(f"LND_GRPC: new_address_impl(input={input})") - - t = 1 if input.type == OnchainAddressType.NP2WKH else 2 - try: - req = ln.NewAddressRequest(type=t) - response = await _lnd_stub.NewAddress(req) - return response.address - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def send_coins_impl(input: SendCoinsInput) -> SendCoinsResponse: - logging.debug(f"LND_GRPC: send_coins_impl(input={input})") - - try: - r = ln.SendCoinsRequest( - addr=input.address, - amount=input.amount, - target_conf=input.target_conf, - sat_per_vbyte=input.sat_per_vbyte, - min_confs=input.min_confs, - label=input.label, - ) - - response = await _lnd_stub.SendCoins(r) - r = SendCoinsResponse.from_lnd_grpc(response, input) - await broadcast_sse_msg(SSE.LN_ONCHAIN_PAYMENT_STATUS, r.dict()) - return r - except grpc.aio._call.AioRpcError as error: - _check_if_locked() - details = error.details() - if details and details.find("invalid bech32 string") > -1: - raise HTTPException( - status.HTTP_400_BAD_REQUEST, - detail="Could not parse destination address, destination should be a valid address.", - ) - elif details and details.find("insufficient funds available") > -1: - raise HTTPException(status.HTTP_412_PRECONDITION_FAILED, detail=details) - else: - raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=details) - - -async def send_payment_impl( - pay_req: str, - timeout_seconds: int, - fee_limit_msat: int, - amount_msat: Optional[int] = None, -) -> Payment: - logging.debug( - f"LND_GRPC: send_payment_impl(pay_req={pay_req}, timeout_seconds={timeout_seconds}, fee_limit_msat={fee_limit_msat}, amount_msat={amount_msat})" - ) - - try: - r = router.SendPaymentRequest( - payment_request=pay_req, - timeout_seconds=timeout_seconds, - fee_limit_msat=fee_limit_msat, - amt_msat=amount_msat, - ) - - p = None - async for response in _router_stub.SendPaymentV2(r): - p = Payment.from_lnd_grpc(response) - await broadcast_sse_msg(SSE.LN_PAYMENT_STATUS, p.dict()) - return p - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - if ( - error.details() != None - and error.details().find("invalid bech32 string") > -1 - ): - raise HTTPException( - status.HTTP_400_BAD_REQUEST, detail="Invalid payment request string" - ) - elif ( - error.details() != None - and error.details().find("OPENSSL_internal:CERTIFICATE_VERIFY_FAILED.") > -1 - ): - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="Invalid LND credentials. SSL certificate verify failed.", - ) - elif ( - error.details() != None - and error.details().find( - "amount must be specified when paying a zero amount invoice" - ) - > -1 - ): - raise HTTPException( - status.HTTP_400_BAD_REQUEST, - detail="amount must be specified when paying a zero amount invoice", - ) - elif ( - error.details() != None - and error.details().find( - "amount must not be specified when paying a non-zero amount invoice" - ) - > -1 - ): - raise HTTPException( - status.HTTP_400_BAD_REQUEST, - detail="amount must not be specified when paying a non-zero amount invoice", - ) - elif ( - error.details() != None - and error.details().find("invoice is already paid") > -1 - ): - raise HTTPException( - status.HTTP_409_CONFLICT, detail="invoice is already paid" - ) - else: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def get_ln_info_impl() -> LnInfo: - logging.debug(f"LND_GRPC: get_ln_info_impl()") - - if not _initialized: - raise HTTPException( - status.HTTP_503_SERVICE_UNAVAILABLE, detail="LND not fully initialized" - ) - - try: - req = ln.GetInfoRequest() - response = await _lnd_stub.GetInfo(req) - return LnInfo.from_lnd_grpc(get_implementation_name(), response) - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def _wait_wallet_fully_ready(): - logging.debug(f"LND_GRPC: _wait_wallet_fully_ready()") - - # This must only be called after unlocking the wallet. - - while True: - try: - info = await _lnd_stub.GetInfo(ln.GetInfoRequest()) - - if info != None: - logging.debug( - f"LND_GRPC: _wait_wallet_fully_ready() breaking out of loop" - ) - break - except grpc.aio._call.AioRpcError as error: - details = error.details() - if ( - "the RPC server is in the process of starting up, but not yet ready to accept calls" - in details - ): - # message from LND AFTER unlocking the wallet - await init_queue.put( - InitLnRepoUpdate( - state=LnInitState.BOOTSTRAPPING_AFTER_UNLOCK, - msg="The RPC server is in the process of starting up, but not yet ready to accept calls", - ) - ) - await asyncio.sleep(0.1) - else: - logging.error(f"LND_GRPC: Unknown error: {details}") - raise - - -async def unlock_wallet_impl(password: str) -> bool: - logging.debug(f"LND_GRPC: unlock_wallet_impl(password=wedontlogpasswords)") - - try: - if _channel is None: - _create_stubs() - - req = unlocker.UnlockWalletRequest(wallet_password=bytes(password, "utf-8")) - await _wallet_unlocker.UnlockWallet(req) - await _wait_wallet_fully_ready() - return True - except grpc.aio._call.AioRpcError as error: - if error.details().find("invalid passphrase") > -1: - raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail=error.details()) - elif error.details().find("wallet already unlocked") > -1: - raise HTTPException( - status.HTTP_412_PRECONDITION_FAILED, detail=error.details() - ) - else: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def listen_invoices() -> Invoice: - logging.debug(f"LND_GRPC: listen_invoices()") - - request = ln.InvoiceSubscription() - try: - async for r in _lnd_stub.SubscribeInvoices(request): - yield Invoice.from_lnd_grpc(r) - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) - - -async def listen_forward_events() -> ForwardSuccessEvent: - logging.debug(f"LND_GRPC: listen_forward_events()") - - request = router.SubscribeHtlcEventsRequest() - try: - _fwd_cache = {} - - async for e in _router_stub.SubscribeHtlcEvents(request): - if e.event_type != 3: - continue - - evt = str(e) - failed_event = "forward_fail_event" in evt or "link_fail_event" in evt - if not e.incoming_htlc_id in _fwd_cache and not failed_event: - _fwd_cache[e.incoming_htlc_id] = e - elif e.incoming_htlc_id in _fwd_cache and not failed_event: - if hasattr(e, "settle_event") and len(e.settle_event.preimage) > 0: - old_e = _fwd_cache[e.incoming_htlc_id] - del _fwd_cache[e.incoming_htlc_id] - amt_in_msat = old_e.forward_event.info.incoming_amt_msat - amt_out_msat = old_e.forward_event.info.outgoing_amt_msat - fee = amt_in_msat - amt_out_msat - yield ForwardSuccessEvent( - timestamp_ns=e.timestamp_ns, - chan_id_in=e.incoming_channel_id, - chan_id_out=e.outgoing_channel_id, - amt_in_msat=amt_in_msat, - amt_out_msat=amt_out_msat, - fee_msat=fee, - ) - elif failed_event and e.incoming_htlc_id in _fwd_cache: - del _fwd_cache[e.incoming_htlc_id] - - except grpc.aio._call.AioRpcError as error: - _check_if_locked(error) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) +from app.repositories.ln_impl.ln_base import LightningNodeBase def _check_if_locked(error): @@ -727,123 +47,803 @@ def _check_if_locked(error): ) -async def channel_open_impl( - local_funding_amount: int, node_URI: str, target_confs: int -) -> str: - logging.debug( - f"LND_GRPC: channel_open_impl(local_funding_amount={local_funding_amount}, node_URI={node_URI}, target_confs={target_confs})" - ) +# Due to updated ECDSA generated tls.cert we need to let gprc know that +# we need to use that cipher suite otherwise there will be a handshake +# error when we communicate with the lnd rpc server. +os.environ["GRPC_SSL_CIPHER_SUITES"] = "HIGH+ECDSA" - try: +# Uncomment to see full gRPC logs +# os.environ["GRPC_TRACE"] = "all" +# os.environ["GRPC_VERBOSITY"] = "DEBUG" - pubkey = node_URI.split("@")[0] - host = node_URI.split("@")[1] - # make sure to be connected to peer - r = ln.ConnectPeerRequest( - addr=ln.LightningAddress(pubkey=pubkey, host=host), - perm=False, - timeout=10, +class LnNodeLNDgRPC(LightningNodeBase): + _lnd_connect_error_debug_msg = """ +LND_GRPC: Unable to connect to LND. Possible reasons: +* Node is not reachable (ports, network down, ...) +* Macaroon is not correct +* IP is not included in LND tls certificate + Add tlsextraip=192.168.1.xxx to lnd.conf and restart LND. + This will recreate the TLS certificate. The .env must be adapted accordingly. +* TLS certificate is wrong. (settings changed, ...) + +To Debug gRPC problems uncomment the following line in app.repositories.ln_impl.lnd_grpc.py +# os.environ["GRPC_VERBOSITY"] = "DEBUG" +This will show more debug information. + """ + + # Decoding the payment request take a long time, + # hence we build a simple cache here. + _memo_cache = {} + _initialized = False + + def _create_stubs(self) -> None: + if self._channel is not None: + logging.warning("LND_GRPC: gRPC channel already created.") + return + + self._channel = grpc.aio.secure_channel( + self._lnd_grpc_url, self._combined_creds ) + self._lnd_stub = lnrpc.LightningStub(self._channel) + self._router_stub = routerrpc.RouterStub(self._channel) + self._wallet_unlocker = unlockerrpc.WalletUnlockerStub(self._channel) + + logging.debug("LND_GRPC: Created LND gRPC stubs") + + def get_implementation_name(self) -> str: + return "LND_GRPC" + + async def _check_lnd_status( + self, + sleep_time: float = 2, + ) -> AsyncGenerator[InitLnRepoUpdate, None]: + logging.debug("LND_GRPC: _check_lnd_status() start") + + self._lnd_connect_error_debug_msg_sent = False + + # Create a temporary channel which will be destroyed at each iteration + # Reason is that gRPC seems to only try and connect every 5 seconds to + # the node if it is not running. To avoid the delay we create a new + # channel each iteration. + + temp_channel = None + temp_stub = None + while True: + try: + if temp_channel is None: + if self._channel is not None: + temp_channel = self._channel + temp_stub = self._lnd_stub + else: + temp_channel = grpc.aio.secure_channel( + self._lnd_grpc_url, self._combined_creds + ) + temp_stub = lnrpc.LightningStub(temp_channel) + await temp_stub.GetInfo(ln.GetInfoRequest()) + + if self._channel is None: + self._create_stubs() + + await self._init_queue.put(InitLnRepoUpdate(state=LnInitState.DONE)) + break + except grpc.aio._call.AioRpcError as error: + details = error.details() + logging.debug(f"LND_GRPC: Waiting for LND daemon... Details {details}") + + if "failed to connect to all addresses" in details: + await self._init_queue.put( + InitLnRepoUpdate( + state=LnInitState.OFFLINE, + msg="Unable to connect to LND daemon, waiting...", + ) + ) + + if not self._lnd_connect_error_debug_msg_sent: + logging.debug(self._lnd_connect_error_debug_msg) + self._lnd_connect_error_debug_msg_sent = True + + await temp_channel.close() + temp_channel = None + elif "waiting to start, RPC services not available" in details: + await self._init_queue.put( + InitLnRepoUpdate( + state=LnInitState.BOOTSTRAPPING, + msg="Connected but waiting to start, RPC services not available", + ) + ) + await temp_channel.close() + temp_channel = None + elif "wallet locked, unlock it to enable full RPC access" in details: + await self._init_queue.put( + InitLnRepoUpdate( + state=LnInitState.LOCKED, + msg="Wallet locked, unlock it to enable full RPC access", + ) + ) + if temp_channel != self._channel: + await temp_channel.close() + temp_channel = None + elif ( + "the RPC server is in the process of starting up, but not yet ready to accept calls" + in details + ): + # message from LND AFTER unlocking the wallet + await self._init_queue.put( + InitLnRepoUpdate( + state=LnInitState.BOOTSTRAPPING_AFTER_UNLOCK, + msg="The RPC server is in the process of starting up, but not yet ready to accept calls", + ) + ) + else: + logging.error(f"LND_GRPC: Unknown error: {details}") + raise + + logging.debug( + f"LND_GRPC: _check_lnd_status() sleeping {sleep_time} seconds..." + ) + await asyncio.sleep(sleep_time) + + logging.debug("LND_GRPC: _check_lnd_status() done") + + async def initialize(self) -> AsyncGenerator[InitLnRepoUpdate, None]: + logging.debug("LND_GRPC: Unable to connect to LND daemon, waiting...") + + if self._initialized: + logging.warning( + "LND_GRPC: Connection already initialized. This function must not be called twice." + ) + yield InitLnRepoUpdate(state=LnInitState.DONE) + + lnd_macaroon = config_get_hex_str(dconfig("lnd_macaroon"), name="lnd_macaroon") + lnd_cert = bytes.fromhex( + config_get_hex_str(dconfig("lnd_cert"), name="lnd_cert") + ) + + def metadata_callback(context, callback): + # for more info see grpc docs + callback([("macaroon", lnd_macaroon)], None) + + lnd_grpc_ip = dconfig("lnd_grpc_ip") + lnd_grpc_port = dconfig("lnd_grpc_port") + self._lnd_grpc_url = lnd_grpc_ip + ":" + lnd_grpc_port + + auth_creds = grpc.metadata_call_credentials(metadata_callback) + ssl_creds = grpc.ssl_channel_credentials(lnd_cert) + self._combined_creds = grpc.composite_channel_credentials(ssl_creds, auth_creds) + self._channel = None + self._lnd_stub = None + self._router_stub = None + self._wallet_unlocker = None + + self._init_queue = asyncio.Queue() + + logging.info("LND_GRPC: Unable to connect to LND daemon, waiting...") + + loop = asyncio.get_event_loop() + task = loop.create_task(self._check_lnd_status(sleep_time=2)) + + while not self._initialized: + res = await self._init_queue.get() # type: InitLnRepoUpdate + + if ( + res.state == LnInitState.BOOTSTRAPPING_AFTER_UNLOCK + and self._channel is None + ): + task.cancel() + + if self._channel == None: + # if res == _API_WALLET_UNLOCK_EVENT the endpoint function will have + # created the channel for us. + self._create_stubs() + + task = loop.create_task(self._check_lnd_status(sleep_time=0.5)) + elif res.state == LnInitState.DONE: + self._initialized = True + if not task.cancelled(): + task.cancel() + elif ( + res.state == LnInitState.OFFLINE + or res.state == LnInitState.LOCKED + or res.state == LnInitState.BOOTSTRAPPING_AFTER_UNLOCK + ): + pass # do nothing here + else: + logging.warning( + f"LND_GRPC: Unhandled initialization event: {res.dict()}" + ) + + yield res + + logging.info("LND_GRPC: Initialization complete.") + + async def get_wallet_balance(self) -> WalletBalance: + logging.debug("LND_GRPC: get_wallet_balance() ") + try: - await _lnd_stub.ConnectPeer(r) + w_req = ln.WalletBalanceRequest() + onchain = await self._lnd_stub.WalletBalance(w_req) + + c_req = ln.ChannelBalanceRequest() + channel = await self._lnd_stub.ChannelBalance(c_req) + + return WalletBalance.from_lnd_grpc(onchain, channel) except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def list_all_tx( + self, successful_only: bool, index_offset: int, max_tx: int, reversed: bool + ) -> List[GenericTx]: + logging.debug( + f"LND_GRPC: list_all_tx(successful_only={successful_only}, index_offset={index_offset}, max_tx={max_tx}, reversed={reversed})" + ) + + # TODO: find a better caching strategy + list_invoice_req = ln.ListInvoiceRequest( + pending_only=successful_only, + index_offset=0, + num_max_invoices=0, + reversed=reversed, + ) + + get_tx_req = ln.GetTransactionsRequest() + + list_payments_req = ln.ListPaymentsRequest( + include_incomplete=not successful_only, + index_offset=0, + max_payments=0, + reversed=reversed, + ) + + try: + res = await asyncio.gather( + *[ + self._lnd_stub.ListInvoices(list_invoice_req), + self._lnd_stub.GetTransactions(get_tx_req), + self._lnd_stub.ListPayments(list_payments_req), + ] + ) + + tx = [] + for i in res[0].invoices: + tx.append(GenericTx.from_lnd_grpc_invoice(i)) + for t in res[1].transactions: + tx.append(GenericTx.from_lnd_grpc_onchain_tx(t)) + for p in res[2].payments: + comment = "" + if p.payment_request in self._memo_cache: + comment = self._memo_cache[p.payment_request] + else: + if p.payment_request is not None and p.payment_request != "": + pr = await self.decode_pay_request(p.payment_request) + comment = pr.description + self._memo_cache[p.payment_request] = pr.description + tx.append(GenericTx.from_lnd_grpc_payment(p, comment)) + + def sortKey(e: GenericTx): + return e.time_stamp + + tx.sort(key=sortKey) + + if reversed: + tx.reverse() + + l = len(tx) + for i in range(l): + tx[i].index = i + + if max_tx == 0: + max_tx = l + + return tx[index_offset : index_offset + max_tx] + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def list_invoices( + self, + pending_only: bool, + index_offset: int, + num_max_invoices: int, + reversed: bool, + ): + logging.debug("LND_GRPC: list_invoices() ") + + try: + req = ln.ListInvoiceRequest( + pending_only=pending_only, + index_offset=index_offset, + num_max_invoices=num_max_invoices, + reversed=reversed, + ) + response = await self._lnd_stub.ListInvoices(req) + return [Invoice.from_lnd_grpc(i) for i in response.invoices] + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def list_on_chain_tx(self) -> List[OnChainTransaction]: + logging.debug("LND_GRPC: list_on_chain_tx() ") + + try: + req = ln.GetTransactionsRequest() + response = await self._lnd_stub.GetTransactions(req) + return [OnChainTransaction.from_lnd_grpc(t) for t in response.transactions] + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def list_payments( + self, + include_incomplete: bool, + index_offset: int, + max_payments: int, + reversed: bool, + ): + logging.debug( + f"LND_GRPC: list_payments(include_incomplete={include_incomplete}, index_offset{index_offset}, max_payments={max_payments}, reversed={reversed})" + ) + + try: + req = ln.ListPaymentsRequest( + include_incomplete=include_incomplete, + index_offset=index_offset, + max_payments=max_payments, + reversed=reversed, + ) + response = await self._lnd_stub.ListPayments(req) + return [Payment.from_lnd_grpc(p) for p in response.payments] + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def add_invoice( + self, + value_msat: int, + memo: str = "", + expiry: int = 3600, + is_keysend: bool = False, + ) -> Invoice: + logging.debug( + f"LND_GRPC: add_invoice(value_msat={value_msat}, memo={memo}, expiry={expiry}, is_keysend={is_keysend})" + ) + + try: + i = ln.Invoice( + memo=memo, + value_msat=value_msat, + expiry=expiry, + is_keysend=is_keysend, + ) + + response = await self._lnd_stub.AddInvoice(i) + + # Can't use Invoice.from_lnd_grpc() here because + # the response is not a standard invoice + invoice = Invoice( + memo=memo, + expiry=expiry, + r_hash=response.r_hash.hex(), + payment_request=response.payment_request, + add_index=response.add_index, + payment_addr=response.payment_addr.hex(), + state=InvoiceState.OPEN, + is_keysend=is_keysend, + value_msat=value_msat, + ) + + return invoice + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def decode_pay_request(self, pay_req: str) -> PaymentRequest: + logging.debug(f"LND_GRPC: decode_pay_request(pay_req={pay_req})") + + try: + req = ln.PayReqString(pay_req=pay_req) + res = await self._lnd_stub.DecodePayReq(req) + return PaymentRequest.from_lnd_grpc(res) + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) if ( error.details() != None - and error.details().find("already connected to peer") > -1 + and error.details().find("checksum failed.") > -1 ): - print("ALREADY CONNECTED TO PEER") - print(str(pubkey)) - + raise HTTPException( + status.HTTP_400_BAD_REQUEST, detail="Invalid payment request string" + ) else: - raise error + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) - # open channel - r = ln.OpenChannelRequest( - node_pubkey=bytes.fromhex(pubkey), - local_funding_amount=local_funding_amount, - target_conf=target_confs, - ) - async for response in _lnd_stub.OpenChannel(r): - # TODO: this is still some bytestring that needs correct conversion to a string txid (ok OK for now) - return str(response.chan_pending.txid.hex()) + async def get_fee_revenue(self) -> FeeRevenue: + logging.debug(f"LND_GRPC: get_fee_revenue()") - except grpc.aio._call.AioRpcError as error: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + req = ln.FeeReportRequest() + res = await self._lnd_stub.FeeReport(req) + return FeeRevenue.from_lnd_grpc(res) + + async def new_address(self, input: NewAddressInput) -> str: + logging.debug(f"LND_GRPC: new_address(input={input})") + + t = 1 if input.type == OnchainAddressType.NP2WKH else 2 + try: + req = ln.NewAddressRequest(type=t) + response = await self._lnd_stub.NewAddress(req) + return response.address + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def send_coins(self, input: SendCoinsInput) -> SendCoinsResponse: + logging.debug(f"LND_GRPC: send_coins(input={input})") + + try: + r = ln.SendCoinsRequest( + addr=input.address, + amount=input.amount, + target_conf=input.target_conf, + sat_per_vbyte=input.sat_per_vbyte, + min_confs=input.min_confs, + label=input.label, + ) + + response = await self._lnd_stub.SendCoins(r) + r = SendCoinsResponse.from_lnd_grpc(response, input) + await broadcast_sse_msg(SSE.LN_ONCHAIN_PAYMENT_STATUS, r.dict()) + return r + except grpc.aio._call.AioRpcError as error: + _check_if_locked() + details = error.details() + if details and details.find("invalid bech32 string") > -1: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + detail="Could not parse destination address, destination should be a valid address.", + ) + elif details and details.find("insufficient funds available") > -1: + raise HTTPException(status.HTTP_412_PRECONDITION_FAILED, detail=details) + else: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=details + ) + + async def send_payment( + self, + pay_req: str, + timeout_seconds: int, + fee_limit_msat: int, + amount_msat: Optional[int] = None, + ) -> Payment: + logging.debug( + f"LND_GRPC: send_payment(pay_req={pay_req}, timeout_seconds={timeout_seconds}, fee_limit_msat={fee_limit_msat}, amount_msat={amount_msat})" ) + try: + r = router.SendPaymentRequest( + payment_request=pay_req, + timeout_seconds=timeout_seconds, + fee_limit_msat=fee_limit_msat, + amt_msat=amount_msat, + ) -async def peer_resolve_alias(node_pub: str) -> str: - logging.debug(f"LND_GRPC: peer_resolve_alias(node_pub={node_pub})") + p = None + async for response in self._router_stub.SendPaymentV2(r): + p = Payment.from_lnd_grpc(response) + await broadcast_sse_msg(SSE.LN_PAYMENT_STATUS, p.dict()) + return p + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + if ( + error.details() != None + and error.details().find("invalid bech32 string") > -1 + ): + raise HTTPException( + status.HTTP_400_BAD_REQUEST, detail="Invalid payment request string" + ) + elif ( + error.details() != None + and error.details().find("OPENSSL_internal:CERTIFICATE_VERIFY_FAILED.") + > -1 + ): + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Invalid LND credentials. SSL certificate verify failed.", + ) + elif ( + error.details() != None + and error.details().find( + "amount must be specified when paying a zero amount invoice" + ) + > -1 + ): + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + detail="amount must be specified when paying a zero amount invoice", + ) + elif ( + error.details() != None + and error.details().find( + "amount must not be specified when paying a non-zero amount invoice" + ) + > -1 + ): + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + detail="amount must not be specified when paying a non-zero amount invoice", + ) + elif ( + error.details() != None + and error.details().find("invoice is already paid") > -1 + ): + raise HTTPException( + status.HTTP_409_CONFLICT, detail="invoice is already paid" + ) + else: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) - # get fresh list of peers and their aliases - try: + async def get_ln_info(self) -> LnInfo: + logging.debug(f"LND_GRPC: get_ln_info()") - request = ln.NodeInfoRequest(pub_key=node_pub, include_channels=False) - response = await _lnd_stub.GetNodeInfo(request) - return str(response.node.alias) + if not self._initialized: + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, detail="LND not fully initialized" + ) - except grpc.aio._call.AioRpcError as error: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + try: + req = ln.GetInfoRequest() + response = await self._lnd_stub.GetInfo(req) + return LnInfo.from_lnd_grpc(self.get_implementation_name(), response) + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def _wait_wallet_fully_ready(self): + logging.debug(f"LND_GRPC: _wait_wallet_fully_ready()") + + # This must only be called after unlocking the wallet. + + while True: + try: + info = await self.__lnd_stub.GetInfo(ln.GetInfoRequest()) + + if info != None: + logging.debug( + f"LND_GRPC: _wait_wallet_fully_ready() breaking out of loop" + ) + break + except grpc.aio._call.AioRpcError as error: + details = error.details() + if ( + "the RPC server is in the process of starting up, but not yet ready to accept calls" + in details + ): + # message from LND AFTER unlocking the wallet + await self._init_queue.put( + InitLnRepoUpdate( + state=LnInitState.BOOTSTRAPPING_AFTER_UNLOCK, + msg="The RPC server is in the process of starting up, but not yet ready to accept calls", + ) + ) + await asyncio.sleep(0.1) + else: + logging.error(f"LND_GRPC: Unknown error: {details}") + raise + + async def unlock_wallet(self, password: str) -> bool: + logging.debug(f"LND_GRPC: unlock_wallet(password=wedontlogpasswords)") + + try: + if self._channel is None: + self._create_stubs() + + req = unlocker.UnlockWalletRequest(wallet_password=bytes(password, "utf-8")) + await self._wallet_unlocker.UnlockWallet(req) + await self._wait_wallet_fully_ready() + return True + except grpc.aio._call.AioRpcError as error: + if error.details().find("invalid passphrase") > -1: + raise HTTPException( + status.HTTP_401_UNAUTHORIZED, detail=error.details() + ) + elif error.details().find("wallet already unlocked") > -1: + raise HTTPException( + status.HTTP_412_PRECONDITION_FAILED, detail=error.details() + ) + else: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def listen_invoices(self) -> AsyncGenerator[Invoice, None]: + logging.debug(f"LND_GRPC: listen_invoices()") + + request = ln.InvoiceSubscription() + try: + async for r in self._lnd_stub.SubscribeInvoices(request): + yield Invoice.from_lnd_grpc(r) + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def listen_forward_events(self) -> ForwardSuccessEvent: + logging.debug(f"LND_GRPC: listen_forward_events()") + + request = router.SubscribeHtlcEventsRequest() + try: + _fwd_cache = {} + + async for e in self._router_stub.SubscribeHtlcEvents(request): + if e.event_type != 3: + continue + + evt = str(e) + failed_event = "forward_fail_event" in evt or "link_fail_event" in evt + if not e.incoming_htlc_id in _fwd_cache and not failed_event: + _fwd_cache[e.incoming_htlc_id] = e + elif e.incoming_htlc_id in _fwd_cache and not failed_event: + if hasattr(e, "settle_event") and len(e.settle_event.preimage) > 0: + old_e = _fwd_cache[e.incoming_htlc_id] + del _fwd_cache[e.incoming_htlc_id] + amt_in_msat = old_e.forward_event.info.incoming_amt_msat + amt_out_msat = old_e.forward_event.info.outgoing_amt_msat + fee = amt_in_msat - amt_out_msat + yield ForwardSuccessEvent( + timestamp_ns=e.timestamp_ns, + chan_id_in=e.incoming_channel_id, + chan_id_out=e.outgoing_channel_id, + amt_in_msat=amt_in_msat, + amt_out_msat=amt_out_msat, + fee_msat=fee, + ) + elif failed_event and e.incoming_htlc_id in _fwd_cache: + del _fwd_cache[e.incoming_htlc_id] + + except grpc.aio._call.AioRpcError as error: + _check_if_locked(error) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def channel_open( + self, local_funding_amount: int, node_URI: str, target_confs: int + ) -> str: + logging.debug( + f"LND_GRPC: channel_open(local_funding_amount={local_funding_amount}, node_URI={node_URI}, target_confs={target_confs})" ) + try: + pubkey = node_URI.split("@")[0] + host = node_URI.split("@")[1] -async def channel_list_impl() -> List[Channel]: - logging.debug(f"LND_GRPC: channel_list_impl()") + # make sure to be connected to peer + r = ln.ConnectPeerRequest( + addr=ln.LightningAddress(pubkey=pubkey, host=host), + perm=False, + timeout=10, + ) + try: + await self._lnd_stub.ConnectPeer(r) + except grpc.aio._call.AioRpcError as error: + if ( + error.details() != None + and error.details().find("already connected to peer") > -1 + ): + print("ALREADY CONNECTED TO PEER") + print(str(pubkey)) - try: + else: + raise error - request = ln.ListChannelsRequest() - response = await _lnd_stub.ListChannels(request) + # open channel + r = ln.OpenChannelRequest( + node_pubkey=bytes.fromhex(pubkey), + local_funding_amount=local_funding_amount, + target_conf=target_confs, + ) + async for response in self._lnd_stub.OpenChannel(r): + # TODO: this is still some bytestring that needs correct conversion to a string txid (ok OK for now) + return str(response.chan_pending.txid.hex()) - channels = [] - for channel_grpc in response.channels: - channel = Channel.from_lnd_grpc(channel_grpc) - channel.peer_alias = await peer_resolve_alias(channel.peer_publickey) - channels.append(channel) + except grpc.aio._call.AioRpcError as error: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) - request = ln.PendingChannelsRequest() - response = await _lnd_stub.PendingChannels(request) - for channel_grpc in response.pending_open_channels: - channel = Channel.from_lnd_grpc_pending(channel_grpc.channel) - channel.peer_alias = await peer_resolve_alias(channel.peer_publickey) - channels.append(channel) + async def peer_resolve_alias(self, node_pub: str) -> str: + logging.debug(f"LND_GRPC: peer_resolve_alias(node_pub={node_pub})") - return channels + # get fresh list of peers and their aliases + try: - except grpc.aio._call.AioRpcError as error: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + request = ln.NodeInfoRequest(pub_key=node_pub, include_channels=False) + response = await self._lnd_stub.GetNodeInfo(request) + return str(response.node.alias) + + except grpc.aio._call.AioRpcError as error: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def channel_list(self) -> List[Channel]: + logging.debug(f"LND_GRPC: channel_list()") + + try: + + request = ln.ListChannelsRequest() + response = await self._lnd_stub.ListChannels(request) + + channels = [] + for channel_grpc in response.channels: + channel = Channel.from_lnd_grpc(channel_grpc) + channel.peer_alias = await self.peer_resolve_alias( + channel.peer_publickey + ) + channels.append(channel) + + request = ln.PendingChannelsRequest() + response = await self._lnd_stub.PendingChannels(request) + for channel_grpc in response.pending_open_channels: + channel = Channel.from_lnd_grpc_pending(channel_grpc.channel) + channel.peer_alias = await self.peer_resolve_alias( + channel.peer_publickey + ) + channels.append(channel) + + return channels + + except grpc.aio._call.AioRpcError as error: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) + + async def channel_close(self, channel_id: int, force_close: bool) -> str: + logging.debug( + f"LND_GRPC: channel_close(channel_id={channel_id}, force_close={force_close})" ) + if not ":" in channel_id: + raise ValueError("channel_id must contain : for lnd") -async def channel_close_impl(channel_id: int, force_close: bool) -> str: - logging.debug( - f"LND_GRPC: channel_close_impl(channel_id={channel_id}, force_close={force_close})" - ) + try: - if not ":" in channel_id: - raise ValueError("channel_id must contain : for lnd") + funding_txid = channel_id.split(":")[0] + output_index = channel_id.split(":")[1] - try: + request = ln.CloseChannelRequest( + channel_point=ln.ChannelPoint( + funding_txid_str=funding_txid, output_index=int(output_index) + ), + force=force_close, + target_conf=6, + ) + async for response in self._lnd_stub.CloseChannel(request): + # TODO: this is still some bytestring that needs correct conversion to a string txid (ok OK for now) + return str(response.close_pending.txid.hex()) - funding_txid = channel_id.split(":")[0] - output_index = channel_id.split(":")[1] - - request = ln.CloseChannelRequest( - channel_point=ln.ChannelPoint( - funding_txid_str=funding_txid, output_index=int(output_index) - ), - force=force_close, - target_conf=6, - ) - async for response in _lnd_stub.CloseChannel(request): - # TODO: this is still some bytestring that needs correct conversion to a string txid (ok OK for now) - return str(response.close_pending.txid.hex()) - - except grpc.aio._call.AioRpcError as error: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() - ) + except grpc.aio._call.AioRpcError as error: + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details() + ) diff --git a/app/repositories/ln_impl/specializations/cln_grpc_blitz.py b/app/repositories/ln_impl/specializations/cln_grpc_blitz.py index 81c9439..1ea3ab9 100644 --- a/app/repositories/ln_impl/specializations/cln_grpc_blitz.py +++ b/app/repositories/ln_impl/specializations/cln_grpc_blitz.py @@ -6,7 +6,6 @@ from decouple import config from fastapi.exceptions import HTTPException from starlette import status -import app.repositories.ln_impl.cln_grpc as cln_main from app.core_utils import call_script2, redis_get from app.models.lightning import ( Channel, @@ -25,307 +24,240 @@ from app.models.lightning import ( SendCoinsResponse, WalletBalance, ) - -# RaspiBlitz implements a lock function on top of CLN, so we need to implement this on Blitz only. +from app.repositories.ln_impl.cln_grpc import LnNodeCLNgRPC -_unlocked = False +class LnNodeCLNgRPCBlitz(LnNodeCLNgRPC): + # RaspiBlitz implements a lock function on top of CLN, so we need to implement this on Blitz only. -_NETWORK = config("network", default="mainnet") + _unlocked = False + _NETWORK = config("network", default="mainnet") -def get_implementation_name() -> str: - return "CLN_GRPC_BLITZ" + def get_implementation_name(self) -> str: + return "CLN_GRPC_BLITZ" + async def initialize(self) -> AsyncGenerator[InitLnRepoUpdate, None]: + logging.debug("CLN_GRPC_BLITZ: RaspiBlitz is locked, waiting for unlock...") -async def initialize_impl() -> AsyncGenerator[InitLnRepoUpdate, None]: - logging.debug("CLN_GRPC_BLITZ: RaspiBlitz is locked, waiting for unlock...") - - global _unlocked - - while not _unlocked: - key = f"ln_cl_{_NETWORK}_locked" - res = await redis_get(key) - if res == "0": - logging.debug( - f"CLN_GRPC_BLITZ: Redis key {key} indicates that RaspiBlitz has been unlocked" - ) - - _unlocked = True - yield InitLnRepoUpdate(state=LnInitState.BOOTSTRAPPING_AFTER_UNLOCK) - break - elif res == "1": - logging.debug( - f"CLN_GRPC_BLITZ: Redis key {key} indicates that RaspiBlitz is still locked" - ) - - yield InitLnRepoUpdate( - state=LnInitState.LOCKED, - msg="Wallet locked, unlock it to enable full RPC access", - ) - else: - logging.error( - f"CLN_GRPC_BLITZ: Redis key {key} returns an unexpected value: {res}" - ) - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Unknown lock status: {res}", - ) - - await asyncio.sleep(2) - - async for u in cln_main.initialize_impl(): - yield u - if u.state == LnInitState.DONE: - break - - logging.info("CLN_GRPC_BLITZ: Initialization complete.") - - -async def get_wallet_balance_impl() -> WalletBalance: - try: - return await cln_main.get_wallet_balance_impl() - except: - _check_if_locked() - raise - - -async def list_all_tx_impl( - successful_only: bool, index_offset: int, max_tx: int, reversed: bool -) -> List[GenericTx]: - try: - return await cln_main.list_all_tx_impl( - successful_only, index_offset, max_tx, reversed - ) - except: - _check_if_locked() - raise - - -async def list_invoices_impl( - pending_only: bool, index_offset: int, num_max_invoices: int, reversed: bool -) -> List[Invoice]: - try: - return await cln_main.list_invoices_impl( - pending_only, index_offset, num_max_invoices, reversed - ) - except: - _check_if_locked() - raise - - -async def list_on_chain_tx_impl() -> List[OnChainTransaction]: - try: - return await cln_main.list_on_chain_tx_impl() - except: - _check_if_locked() - raise - - -async def list_payments_impl( - include_incomplete: bool, index_offset: int, max_payments: int, reversed: bool -): - try: - return await cln_main.list_payments_impl( - include_incomplete, index_offset, max_payments, reversed - ) - except: - _check_if_locked() - raise - - -async def add_invoice_impl( - value_msat: int, memo: str = "", expiry: int = 3600, is_keysend: bool = False -) -> Invoice: - try: - return await cln_main.add_invoice_impl(value_msat, memo, expiry, is_keysend) - except: - _check_if_locked() - raise - - -async def decode_pay_request_impl(pay_req: str) -> PaymentRequest: - try: - return await cln_main.decode_pay_request_impl(pay_req) - except: - _check_if_locked() - raise - - -async def get_fee_revenue_impl() -> FeeRevenue: - try: - return await cln_main.get_fee_revenue_impl() - except: - _check_if_locked() - raise - - -async def new_address_impl(input: NewAddressInput) -> str: - try: - return await cln_main.new_address_impl(input) - except: - _check_if_locked() - raise - - -async def send_coins_impl(input: SendCoinsInput) -> SendCoinsResponse: - try: - return await cln_main.send_coins_impl(input) - except: - _check_if_locked() - raise - - -async def send_payment_impl( - pay_req: str, - timeout_seconds: int, - fee_limit_msat: int, - amount_msat: Optional[int] = None, -) -> Payment: - try: - return await cln_main.send_payment_impl( - pay_req, timeout_seconds, fee_limit_msat, amount_msat - ) - except: - _check_if_locked() - raise - - -async def get_ln_info_impl() -> LnInfo: - try: - # This will return "CLN_GRPC" and not "CLN_GRPC_BLITZ" to - # not to complicate things further. - # res.implementation = get_implementation_name() - return await cln_main.get_ln_info_impl() - except: - _check_if_locked() - raise - - -async def unlock_wallet_impl(password: str) -> bool: - # RaspiBlitz implements a wallet lock functionality on top of CLN, - # so we need to implement this on Blitz only - - # /home/admin/config.scripts/cl.hsmtool.sh unlock mainnet PASSWORD_C - # cl.hsmtool.sh [unlock] - - global _unlocked - key = f"ln_cl_{_NETWORK}_locked" - res = await redis_get(key) - if res == "0": - raise HTTPException( - status.HTTP_412_PRECONDITION_FAILED, detail="wallet already unlocked" - ) - - res = await call_script2( - f"/home/admin/config.scripts/cl.hsmtool.sh unlock {_NETWORK} {password}" - ) - - if res.return_code == 0: - logging.debug( - f"CLN_GRPC_BLITZ: Unlock script successfully called via API. Waiting for Redis {key} to be set." - ) - - # success: exit 0 - INTERVAL = 1 - total_wait_time = 0 - while total_wait_time < 60: + while not self._unlocked: + key = f"ln_cl_{self._NETWORK}_locked" res = await redis_get(key) if res == "0": - _unlocked = True - return True + logging.debug( + f"CLN_GRPC_BLITZ: Redis key {key} indicates that RaspiBlitz has been unlocked" + ) - await asyncio.sleep(INTERVAL) - total_wait_time += INTERVAL + self._unlocked = True + yield InitLnRepoUpdate(state=LnInitState.BOOTSTRAPPING_AFTER_UNLOCK) + break + elif res == "1": + logging.debug( + f"CLN_GRPC_BLITZ: Redis key {key} indicates that RaspiBlitz is still locked" + ) - logging.debug( - f"CLN_GRPC_BLITZ: Unlock script called successfully but redis key {key} indicates that RaspiBlitz is still locked. Stopped watching after polling for 60s for an unlock signal." + yield InitLnRepoUpdate( + state=LnInitState.LOCKED, + msg="Wallet locked, unlock it to enable full RPC access", + ) + else: + logging.error( + f"CLN_GRPC_BLITZ: Redis key {key} returns an unexpected value: {res}" + ) + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Unknown lock status: {res}", + ) + + await asyncio.sleep(2) + + async for u in super().initialize_impl(): + yield u + if u.state == LnInitState.DONE: + break + + logging.info("CLN_GRPC_BLITZ: Initialization complete.") + + async def get_wallet_balance(self) -> WalletBalance: + self._check_if_locked() + return await super().get_wallet_balance() + + async def list_all_tx( + self, successful_only: bool, index_offset: int, max_tx: int, reversed: bool + ) -> List[GenericTx]: + self._check_if_locked() + return await super().list_all_tx( + successful_only, index_offset, max_tx, reversed ) + async def list_invoices( + self, + pending_only: bool, + index_offset: int, + num_max_invoices: int, + reversed: bool, + ): + self._check_if_locked() + return await super().list_invoices( + pending_only, index_offset, num_max_invoices, reversed + ) + + async def list_on_chain_tx(self) -> List[OnChainTransaction]: + self._check_if_locked() + return await super().list_on_chain_tx() + + async def list_payments( + self, + include_incomplete: bool, + index_offset: int, + max_payments: int, + reversed: bool, + ): + self._check_if_locked() + return await super().list_payments( + include_incomplete, index_offset, max_payments, reversed + ) + + async def add_invoice( + self, + value_msat: int, + memo: str = "", + expiry: int = 3600, + is_keysend: bool = False, + ) -> Invoice: + self._check_if_locked() + return await super().add_invoice(value_msat, memo, expiry, is_keysend) + + async def decode_pay_request(self, pay_req: str) -> PaymentRequest: + self._check_if_locked() + return await super().decode_pay_request(pay_req) + + async def get_fee_revenue(self) -> FeeRevenue: + self._check_if_locked() + return await super().get_fee_revenue() + + async def new_address(self, input: NewAddressInput) -> str: + self._check_if_locked() + return await super().new_address(input) + + async def send_coins(self, input: SendCoinsInput) -> SendCoinsResponse: + self._check_if_locked() + return await super().send_coins(input) + + async def send_payment( + self, + pay_req: str, + timeout_seconds: int, + fee_limit_msat: int, + amount_msat: Optional[int] = None, + ) -> Payment: + self._check_if_locked() + return await super().send_payment( + pay_req, timeout_seconds, fee_limit_msat, amount_msat + ) + + async def get_ln_info(self) -> LnInfo: + self._check_if_locked() + return await super().get_ln_info() + + async def unlock_wallet(self, password: str) -> bool: + # RaspiBlitz implements a wallet lock functionality on top of CLN, + # so we need to implement this on Blitz only + + # /home/admin/config.scripts/cl.hsmtool.sh unlock mainnet PASSWORD_C + # cl.hsmtool.sh [unlock] + + key = f"ln_cl_{self._NETWORK}_locked" + res = await redis_get(key) + if res == "0": + raise HTTPException( + status.HTTP_412_PRECONDITION_FAILED, detail="wallet already unlocked" + ) + + res = await call_script2( + f"/home/admin/config.scripts/cl.hsmtool.sh unlock {self._NETWORK} {password}" + ) + + if res.return_code == 0: + logging.debug( + f"CLN_GRPC_BLITZ: Unlock script successfully called via API. Waiting for Redis {key} to be set." + ) + + # success: exit 0 + INTERVAL = 1 + total_wait_time = 0 + while total_wait_time < 60: + res = await redis_get(key) + if res == "0": + _unlocked = True + return True + + await asyncio.sleep(INTERVAL) + total_wait_time += INTERVAL + + logging.debug( + f"CLN_GRPC_BLITZ: Unlock script called successfully but redis key {key} indicates that RaspiBlitz is still locked. Stopped watching after polling for 60s for an unlock signal." + ) + + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Unknown error while trying to unlock.", + ) + elif res.return_code == 1: + logging.error("CLN_GRPC_BLITZ: Unknown error while trying to unlock.") + logging.error(f"CLN_GRPC_BLITZ: {res.__str__()}") + + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Unknown error while trying to unlock. See the API logs for more info.", + ) + elif res.return_code == 2: + # wrong password: exit 2 + raise HTTPException( + status.HTTP_401_UNAUTHORIZED, detail="invalid passphrase" + ) + elif res.return_code == 3: + # fail to unlock after 1 minute + show logs: exit 3 + raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=res) + raise HTTPException( status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="Unknown error while trying to unlock.", + detail=f"Unknown error while trying to unlock.\n{res}", ) - elif res.return_code == 1: - logging.error("CLN_GRPC_BLITZ: Unknown error while trying to unlock.") - logging.error(f"CLN_GRPC_BLITZ: {res.__str__()}") - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Unknown error while trying to unlock. See the API logs for more info.", - ) - elif res.return_code == 2: - # wrong password: exit 2 - raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="invalid passphrase") - elif res.return_code == 3: - # fail to unlock after 1 minute + show logs: exit 3 - raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=res) - - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Unknown error while trying to unlock.\n{res}", - ) - - -async def listen_invoices() -> AsyncGenerator[Invoice, None]: - try: - async for i in cln_main.listen_invoices(): + async def listen_invoices(self) -> AsyncGenerator[Invoice, None]: + self._check_if_locked() + async for i in super().listen_invoices(): yield i - except: - _check_if_locked() - raise + async def listen_forward_events(self) -> ForwardSuccessEvent: + self._check_if_locked() + async for i in super().listen_forward_events(): + yield i -async def listen_forward_events() -> ForwardSuccessEvent: - try: - async for e in cln_main.listen_forward_events(): - yield e - except: - _check_if_locked() - raise + async def channel_open( + self, local_funding_amount: int, node_URI: str, target_confs: int + ) -> str: + self._check_if_locked() + return await super().channel_open(local_funding_amount, node_URI, target_confs) + async def peer_resolve_alias(self, node_pub: str) -> str: + self._check_if_locked() + return await super().peer_resolve_alias(node_pub) -async def connect_peer_impl(node_URI: str) -> bool: - try: - return await cln_main.connect_peer_impl(node_URI) - except: - _check_if_locked() - raise + async def channel_list(self) -> List[Channel]: + self._check_if_locked() + return await super().channel_list() + async def channel_close(self, channel_id: int, force_close: bool) -> str: + self._check_if_locked() + return await super().channel_close(channel_id, force_close) -async def channel_open_impl( - local_funding_amount: int, node_URI: str, target_confs: int -) -> str: - try: - return await cln_main.channel_open_impl( - local_funding_amount, node_URI, target_confs - ) - except: - _check_if_locked() - raise + def _check_if_locked(self): + logging.debug(f"CLN_GRPC_BLITZ: _check_if_locked()") - -async def channel_list_impl() -> List[Channel]: - try: - return await cln_main.channel_list_impl() - except: - _check_if_locked() - raise - - -async def channel_close_impl(channel_id: int, force_close: bool) -> str: - try: - return await cln_main.channel_close_impl(channel_id, force_close) - except: - _check_if_locked() - raise - - -def _check_if_locked(): - logging.debug(f"CLN_GRPC_BLITZ: _check_if_locked()") - - if not _unlocked: - raise HTTPException( - status.HTTP_423_LOCKED, - detail="Wallet is locked. Unlock via /lightning/unlock-wallet", - ) + if not self._unlocked: + raise HTTPException( + status.HTTP_423_LOCKED, + detail="Wallet is locked. Unlock via /lightning/unlock-wallet", + )