lnd.py (5833B)
1 import subprocess 2 import time 3 import os 4 import json 5 from base64 import b64decode 6 from google.protobuf.json_format import MessageToJson 7 import qrcode 8 import logging 9 10 import config 11 12 13 class lnd: 14 def __init__(self, node_config): 15 from lndgrpc import LNDClient 16 17 self.config = node_config 18 self.is_onchain = False 19 20 # Copy admin macaroon and tls cert to local machine 21 self.copy_certs() 22 23 # Conect to lightning node 24 connection_str = "{}:{}".format(config.host, self.config['lnd_rpcport']) 25 logging.info( 26 "Attempting to connect to lightning node {}. This may take a few seconds...".format( 27 connection_str 28 ) 29 ) 30 31 for i in range(config.connection_attempts): 32 try: 33 logging.info("Attempting to initialise lnd rpc client...") 34 time.sleep(3) 35 self.lnd = LNDClient( 36 "{}:{}".format(config.host, self.config['lnd_rpcport']), 37 macaroon_filepath=self.certs["macaroon"], 38 cert_filepath=self.certs["tls"], 39 ) 40 41 if "invoice" in self.certs["macaroon"]: 42 logging.info("Testing we can fetch invoices...") 43 inv, _ = self.create_lnd_invoice(1) 44 logging.info(inv) 45 else: 46 logging.info("Getting lnd info...") 47 info = self.get_info() 48 logging.info(info) 49 50 logging.info("Successfully contacted lnd.") 51 break 52 53 except Exception as e: 54 logging.error(e) 55 if i < 5: 56 time.sleep(2) 57 else: 58 time.sleep(60) 59 logging.info( 60 "Attempting again... {}/{}...".format( 61 i + 1, config.connection_attempts 62 ) 63 ) 64 else: 65 raise Exception( 66 "Could not connect to lnd. Check your gRPC / port tunneling settings and try again." 67 ) 68 69 logging.info("Ready for payments requests.") 70 return 71 72 def create_qr(self, uuid, address, value): 73 qr_str = "{}".format(address.upper()) 74 img = qrcode.make(qr_str) 75 img.save("static/qr_codes/{}.png".format(uuid)) 76 return 77 78 # Copy tls and macaroon certs from remote machine. 79 def copy_certs(self): 80 self.certs = {"tls": "tls.cert", "macaroon": self.config['lnd_macaroon']} 81 82 if (not os.path.isfile("tls.cert")) or ( 83 not os.path.isfile(self.config['lnd_macaroon']) 84 ): 85 try: 86 tls_file = os.path.join(self.config['lnd_dir'], "tls.cert") 87 macaroon_file = os.path.join( 88 self.config['lnd_dir'], 89 "data/chain/bitcoin/mainnet/{}".format(self.config['lnd_macaroon']), 90 ) 91 92 # SSH copy 93 if config.tunnel_host is not None: 94 logging.warning( 95 "Could not find tls.cert or {} in SatSale folder. \ 96 Attempting to download from remote lnd directory.".format( 97 self.config['lnd_macaroon'] 98 ) 99 ) 100 101 subprocess.run( 102 ["scp", "{}:{}".format(config.tunnel_host, tls_file), "."] 103 ) 104 subprocess.run( 105 [ 106 "scp", 107 "-r", 108 "{}:{}".format(config.tunnel_host, macaroon_file), 109 ".", 110 ] 111 ) 112 113 else: 114 self.certs = { 115 "tls": os.path.expanduser(tls_file), 116 "macaroon": os.path.expanduser(macaroon_file), 117 } 118 119 except Exception as e: 120 logging.error(e) 121 logging.error("Failed to copy tls and macaroon files to local machine.") 122 else: 123 logging.info("Found tls.cert and admin.macaroon.") 124 return 125 126 # Create lightning invoice 127 def create_lnd_invoice(self, btc_amount, memo=None, description_hash=None, expiry=3600): 128 # Multiplying by 10^8 to convert to satoshi units 129 sats_amount = int(float(btc_amount) * 10 ** 8) 130 res = self.lnd.add_invoice( 131 value=sats_amount, memo=memo, description_hash=description_hash, expiry=expiry 132 ) 133 lnd_invoice = json.loads(MessageToJson(res)) 134 135 return lnd_invoice["paymentRequest"], lnd_invoice["rHash"] 136 137 def get_address(self, amount, label, expiry): 138 address, r_hash = self.create_lnd_invoice( 139 amount, memo=label, expiry=expiry) 140 return address, r_hash 141 142 def pay_invoice(self, invoice): 143 ret = json.loads( 144 MessageToJson(self.lnd.send_payment(invoice, fee_limit_msat=20 * 1000)) 145 ) 146 logging.info(ret) 147 return 148 149 def get_info(self): 150 return json.loads(MessageToJson(self.lnd.get_info())) 151 152 def get_uri(self): 153 info = self.get_info() 154 return info["uris"][0] 155 156 # Check whether the payment has been paid 157 def check_payment(self, rhash): 158 invoice_status = json.loads( 159 MessageToJson(self.lnd.lookup_invoice(r_hash_str=b64decode(rhash).hex())) 160 ) 161 162 if "amtPaidSat" not in invoice_status.keys(): 163 conf_paid = 0 164 unconf_paid = 0 165 else: 166 # Store amount paid and convert to BTC units 167 conf_paid = (int(invoice_status["amtPaidSat"]) + 1) / (10 ** 8) 168 unconf_paid = 0 169 170 return conf_paid, unconf_paid