electrum

Electrum Bitcoin wallet
git clone https://git.parazyd.org/electrum
Log | Files | Refs | Submodules

bip39_recovery.py (3044B)

      1 # Copyright (C) 2020 The Electrum developers
      2 # Distributed under the MIT software license, see the accompanying
      3 # file LICENCE or http://www.opensource.org/licenses/mit-license.php
      4 
      5 from typing import TYPE_CHECKING
      6 
      7 from aiorpcx import TaskGroup
      8 
      9 from . import bitcoin
     10 from .constants import BIP39_WALLET_FORMATS
     11 from .bip32 import BIP32_PRIME, BIP32Node
     12 from .bip32 import convert_bip32_path_to_list_of_uint32 as bip32_str_to_ints
     13 from .bip32 import convert_bip32_intpath_to_strpath as bip32_ints_to_str
     14 
     15 if TYPE_CHECKING:
     16     from .network import Network
     17 
     18 
     19 async def account_discovery(network: 'Network', get_account_xpub):
     20     async with TaskGroup() as group:
     21         account_scan_tasks = []
     22         for wallet_format in BIP39_WALLET_FORMATS:
     23             account_scan = scan_for_active_accounts(network, get_account_xpub, wallet_format)
     24             account_scan_tasks.append(await group.spawn(account_scan))
     25     active_accounts = []
     26     for task in account_scan_tasks:
     27         active_accounts.extend(task.result())
     28     return active_accounts
     29 
     30 
     31 async def scan_for_active_accounts(network: 'Network', get_account_xpub, wallet_format):
     32     active_accounts = []
     33     account_path = bip32_str_to_ints(wallet_format["derivation_path"])
     34     while True:
     35         account_xpub = get_account_xpub(account_path)
     36         account_node = BIP32Node.from_xkey(account_xpub)
     37         has_history = await account_has_history(network, account_node, wallet_format["script_type"])
     38         if has_history:
     39             account = format_account(wallet_format, account_path)
     40             active_accounts.append(account)
     41         if not has_history or not wallet_format["iterate_accounts"]:
     42             break
     43         account_path[-1] = account_path[-1] + 1
     44     return active_accounts
     45 
     46 
     47 async def account_has_history(network: 'Network', account_node: BIP32Node, script_type: str) -> bool:
     48     gap_limit = 20
     49     async with TaskGroup() as group:
     50         get_history_tasks = []
     51         for address_index in range(gap_limit):
     52             address_node = account_node.subkey_at_public_derivation("0/" + str(address_index))
     53             pubkey = address_node.eckey.get_public_key_hex()
     54             address = bitcoin.pubkey_to_address(script_type, pubkey)
     55             script = bitcoin.address_to_script(address)
     56             scripthash = bitcoin.script_to_scripthash(script)
     57             get_history = network.get_history_for_scripthash(scripthash)
     58             get_history_tasks.append(await group.spawn(get_history))
     59     for task in get_history_tasks:
     60         history = task.result()
     61         if len(history) > 0:
     62             return True
     63     return False
     64 
     65 
     66 def format_account(wallet_format, account_path):
     67     description = wallet_format["description"]
     68     if wallet_format["iterate_accounts"]:
     69         account_index = account_path[-1] % BIP32_PRIME
     70         description = f'{description} (Account {account_index})'
     71     return {
     72         "description": description,
     73         "derivation_path": bip32_ints_to_str(account_path),
     74         "script_type": wallet_format["script_type"],
     75     }