tbip39_recovery.py - electrum - Electrum Bitcoin wallet
HTML git clone https://git.parazyd.org/electrum
DIR Log
DIR Files
DIR Refs
DIR Submodules
---
tbip39_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 }