xpub.py (3407B)
1 import logging 2 import qrcode 3 import requests 4 import sys 5 import time 6 from bip_utils import Bip84, Bip44Changes, Bip84Coins, Bip44, Bip44Coins 7 8 from utils import btc_amount_format 9 from payments import database 10 11 12 class xpub: 13 def __init__(self, node_config): 14 self.is_onchain = True 15 self.config = node_config 16 self.api = "https://mempool.space/api" 17 18 if "pytest" not in sys.modules: 19 next_n = self.get_next_address_index(self.config["xpub"]) 20 # Warning will be printed for production runs, but not when running tests. 21 if next_n == 0: 22 logging.info( 23 "Deriving addresses for first time from xpub: {}".format( 24 self.config["xpub"] 25 ) 26 ) 27 logging.warn( 28 "YOU MUST CHECK THIS MATCHES THE FIRST ADDRESS IN YOUR WALLET:" 29 ) 30 logging.warn(self.get_address_at_index(next_n)) 31 time.sleep(10) 32 33 logging.info("Fetching blockchain info from {}".format(self.api)) 34 logging.info("Next address shown to users is #{}".format(next_n)) 35 36 def create_qr(self, uuid, address, value): 37 qr_str = "bitcoin:{}?amount={}&label={}".format( 38 address, btc_amount_format(value), uuid 39 ) 40 41 img = qrcode.make(qr_str) 42 img.save("static/qr_codes/{}.png".format(uuid)) 43 return 44 45 def check_payment(self, address, slow=True): 46 conf_paid, unconf_paid = 0, 0 47 try: 48 r = requests.get(self.api + "/address/{}".format(address)) 49 r.raise_for_status() 50 stats = r.json() 51 conf_paid = stats["chain_stats"]["funded_txo_sum"] / (10 ** 8) 52 unconf_paid = stats["mempool_stats"]["funded_txo_sum"] / (10 ** 8) 53 54 # Don't request too often 55 if slow and (conf_paid == 0): 56 time.sleep(1) 57 58 return conf_paid, unconf_paid 59 60 except Exception as e: 61 logging.error( 62 "Failed to fetch address information from mempool: {}".format(e) 63 ) 64 65 return 0, 0 66 67 def get_next_address_index(self, xpub): 68 n = database.get_next_address_index(xpub) 69 return n 70 71 def get_address_at_index(self, index): 72 if self.config["bip"] == "BIP84": 73 bip84_acc = Bip84.FromExtendedKey(self.config["xpub"], Bip84Coins.BITCOIN) 74 child_key = bip84_acc.Change(Bip44Changes.CHAIN_EXT).AddressIndex(index) 75 elif self.config["bip"] == "BIP44": 76 bip44_acc = Bip44.FromExtendedKey(self.config["xpub"], Bip44Coins.BITCOIN) 77 child_key = bip44_acc.Change(Bip44Changes.CHAIN_EXT).AddressIndex(index) 78 else: 79 raise NotImplementedError( 80 "{} is not yet implemented!".format(self.config["bip"]) 81 ) 82 83 address = child_key.PublicKey().ToAddress() 84 return address 85 86 def get_address(self, amount, label, expiry): 87 while True: 88 n = self.get_next_address_index(self.config["xpub"]) 89 address = self.get_address_at_index(n) 90 database.add_generated_address(n, address, self.config["xpub"]) 91 conf_paid, unconf_paid = self.check_payment(address, slow=False) 92 if conf_paid == 0 and unconf_paid == 0: 93 break 94 return address, None