mirror of
https://github.com/cryptoadvance/specter-desktop.git
synced 2026-08-13 12:33:29 +02:00
99f5404705
A feature for a notification system has been added, including websockets and SSL support, improved error handling, and better logging. There have been various bug fixes, refactoring, and documentation updates throughout the process. The feature is now complete and includes new functionality for managing and displaying notifications, as well as testing and debugging tools.m> Co-authored-by: relativisticelectron <60378539+relativisticelectron@users.noreply.github.com>
724 lines
26 KiB
Python
724 lines
26 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from cryptoadvance.specter.devices.bitcoin_core import BitcoinCore
|
|
from cryptoadvance.specter.devices.generic import GenericDevice
|
|
from cryptoadvance.specter.helpers import is_testnet
|
|
from cryptoadvance.specter.process_controller.bitcoind_controller import (
|
|
BitcoindPlainController,
|
|
)
|
|
from cryptoadvance.specter.util.mnemonic import generate_mnemonic
|
|
from cryptoadvance.specter.key import Key
|
|
from cryptoadvance.specter.managers.wallet_manager import WalletManager
|
|
from cryptoadvance.specter.specter_error import SpecterError
|
|
from cryptoadvance.specter.util.descriptor import AddChecksum, Descriptor
|
|
from cryptoadvance.specter.util.wallet_importer import WalletImporter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@patch("cryptoadvance.specter.util.wallet_importer.flash", print)
|
|
@pytest.mark.slow
|
|
def test_WalletManager(
|
|
request,
|
|
devices_filled_data_folder,
|
|
device_manager,
|
|
bitcoin_regtest,
|
|
node,
|
|
node_with_empty_datadir,
|
|
):
|
|
wm = WalletManager(
|
|
devices_filled_data_folder,
|
|
bitcoin_regtest.get_rpc(),
|
|
"regtest",
|
|
device_manager,
|
|
)
|
|
assert wm.rpc_path == "specter"
|
|
# A wallet-creation needs a device
|
|
device = device_manager.get_by_alias("trezor")
|
|
assert device != None
|
|
# Lets's create a wallet with the WalletManager
|
|
wm.create_wallet("a_test_wallet", 1, "wpkh", [device.keys[5]], [device])
|
|
# The wallet-name gets its filename and therefore its alias
|
|
wallet = wm.wallets["a_test_wallet"]
|
|
assert wallet != None
|
|
assert wallet.balance["trusted"] == 0
|
|
assert wallet.balance["untrusted_pending"] == 0
|
|
# this is a sum of both
|
|
assert wallet.amount_total == 0
|
|
address = wallet.getnewaddress()
|
|
bitcoin_regtest.testcoin_faucet(address, amount=3)
|
|
# update the balance
|
|
wallet.update_balance()
|
|
assert wallet.amount_total == 3
|
|
assert wallet.amount_immature == 0
|
|
|
|
# You can create a multisig wallet with the wallet manager like this
|
|
second_device = device_manager.get_by_alias("specter")
|
|
multisig_wallet = wm.create_wallet(
|
|
"a_multisig_test_wallet",
|
|
1,
|
|
"wsh",
|
|
[device.keys[7], second_device.keys[0]],
|
|
[device, second_device],
|
|
)
|
|
|
|
assert len(wm.wallets) == 2
|
|
assert multisig_wallet != None
|
|
assert multisig_wallet.amount_total == 0
|
|
multisig_address = multisig_wallet.getnewaddress()
|
|
bitcoin_regtest.testcoin_faucet(multisig_address, amount=4)
|
|
# update balance
|
|
multisig_wallet.update_balance()
|
|
assert multisig_wallet.amount_total == 4
|
|
# The WalletManager also has a `wallets_names` property, returning a sorted list of the names of all wallets
|
|
assert wm.wallets_names == ["a_multisig_test_wallet", "a_test_wallet"]
|
|
|
|
# You can rename a wallet using the wallet manager using `rename_wallet`, passing the wallet object and the new name to assign to it
|
|
wm.rename_wallet(multisig_wallet, "new_name_test_wallet")
|
|
assert multisig_wallet.name == "new_name_test_wallet"
|
|
assert wm.wallets_names == ["a_test_wallet", "new_name_test_wallet"]
|
|
|
|
# You can also delete a wallet by passing it to the wallet manager's `delete_wallet` method
|
|
# It will delete the json and attempt to remove it from Bitcoin Core
|
|
wallet_fullpath = multisig_wallet.fullpath
|
|
assert os.path.exists(wallet_fullpath)
|
|
# This deletion should also remove the wallet file on the node
|
|
assert wm.delete_wallet(multisig_wallet, node) == (True, True)
|
|
assert len(wm.wallets) == 1
|
|
assert not os.path.exists(wallet_fullpath)
|
|
# Let's artificially unload the wallet in Core before trying to delete it via the Wallet Manager, should raise a SpecterError
|
|
wallet_rpc_path = os.path.join(wm.rpc_path, wallet.alias)
|
|
wm.rpc.unloadwallet(wallet_rpc_path)
|
|
with pytest.raises(
|
|
SpecterError,
|
|
match="Unable to unload the wallet on the node. Aborting the deletion of the wallet ...",
|
|
):
|
|
wm.delete_wallet(wallet, node)
|
|
# Check that the wallet wasn't deleted in Specter because of the RpcError
|
|
assert wm.wallets_names == ["a_test_wallet"]
|
|
# The following deletion should not remove the wallet file on the node
|
|
assert node_with_empty_datadir.datadir == ""
|
|
wm.rpc.loadwallet(wallet_rpc_path) # we need to load the wallet again
|
|
assert wm.delete_wallet(wallet, node_with_empty_datadir) == (True, False)
|
|
assert len(wm.wallets) == 0
|
|
# The wallet in Specter was already deleted, so trying to delete it again should raise a SpecterError
|
|
wm.rpc.loadwallet(wallet_rpc_path) # we need to load the wallet again
|
|
with pytest.raises(
|
|
SpecterError, match="The wallet a_test_wallet has already been deleted."
|
|
):
|
|
assert wm.delete_wallet(wallet)
|
|
|
|
|
|
@pytest.mark.slow
|
|
@pytest.mark.bottleneck
|
|
@pytest.mark.threading
|
|
def test_WalletManager_2_nodes(
|
|
request,
|
|
devices_filled_data_folder,
|
|
device_manager,
|
|
bitcoin_regtest,
|
|
bitcoin_regtest2: BitcoindPlainController,
|
|
caplog,
|
|
):
|
|
caplog.set_level(logging.INFO)
|
|
wm = WalletManager(
|
|
devices_filled_data_folder,
|
|
bitcoin_regtest.get_rpc(),
|
|
"regtest",
|
|
device_manager,
|
|
allow_threading_for_testing=True,
|
|
)
|
|
# Wallet creation needs a device
|
|
device = device_manager.get_by_alias("trezor")
|
|
assert device != None
|
|
first_wallet = wm.create_wallet(
|
|
"a_test_wallet", 1, "wpkh", [device.keys[5]], [device]
|
|
)
|
|
assert wm.wallets_names == ["a_test_wallet"]
|
|
assert wm.chain == "regtest"
|
|
assert wm.working_folder.endswith("regtest")
|
|
assert wm.rpc.port == 18543
|
|
# Change the rpc - this works differently with a different chain!
|
|
# If we use something different that regtest, unfortunately a liquid address gets
|
|
# generated.
|
|
# So we don't test that scanrio here of different chains. We test the scenario with the same chain.
|
|
# but different node
|
|
wm.update(rpc=bitcoin_regtest2.get_rpc(), chain="regtest")
|
|
# A WalletManager uses the chain as an index
|
|
assert list(wm.rpcs.keys()) == [
|
|
"regtest",
|
|
] # wm.rpcs looks like this: {'regtest': <BitcoinRpc http://localhost:18543>, 'regtest2': <BitcoinRpc http://localhost:18544>}
|
|
assert wm.rpc.port == 18544
|
|
assert wm.wallets_names == ["a_test_wallet"]
|
|
assert wm.chain == "regtest"
|
|
assert wm.working_folder.endswith("test")
|
|
second_wallet = wm.create_wallet(
|
|
"a_regtest2_test_wallet", 1, "wpkh", [device.keys[5]], [device]
|
|
)
|
|
# Note: "regtest2" is recognised by the get_network() from embit as Liquid, that is why there is an error in the logs saying the Bitcoin address is not valid since a Liquid address is derived.
|
|
assert len(wm.wallets_names) == 2
|
|
assert wm.wallets_names == ["a_regtest2_test_wallet", "a_test_wallet"]
|
|
|
|
|
|
def test_WalletManager_check_duplicate_keys(empty_data_folder):
|
|
wm = WalletManager(
|
|
empty_data_folder,
|
|
MagicMock(), # needs rpc
|
|
"regtest",
|
|
None,
|
|
)
|
|
key1 = Key(
|
|
"[f3e6eaff/84h/0h/0h]xpub6C5cCQfycZrPJnNg6cDdUU5efJrab8thRQDBxSSB4gP2J3xGdWu8cqiLvPZkejtuaY9LursCn6Es9PqHgLhBktW8217BomGDVBAJjUms8iG",
|
|
"f3e6eaff",
|
|
"84h/0h/0h",
|
|
"",
|
|
None,
|
|
"xpub6C5cCQfycZrPJnNg6cDdUU5efJrab8thRQDBxSSB4gP2J3xGdWu8cqiLvPZkejtuaY9LursCn6Es9PqHgLhBktW8217BomGDVBAJjUms8iG",
|
|
)
|
|
key2 = Key(
|
|
"[1ef4e492/49h/0h/0h]xpub6CRWp2zfwRYsVTuT2p96hKE2UT4vjq9gwvW732KWQjwoG7v6NCXyaTdz7NE5yDxsd72rAGK7qrjF4YVrfZervsJBjsXxvTL98Yhc7poBk7K",
|
|
"1ef4e492",
|
|
"m/49h/0h/0h",
|
|
"sh-wpkh",
|
|
None,
|
|
"xpub6CRWp2zfwRYsVTuT2p96hKE2UT4vjq9gwvW732KWQjwoG7v6NCXyaTdz7NE5yDxsd72rAGK7qrjF4YVrfZervsJBjsXxvTL98Yhc7poBk7K",
|
|
)
|
|
key3 = Key(
|
|
"[1ef4e492/49h/0h/0h]zpub6qk8ok1ouvwM1NkumKnsteGf1F9UUNshFdFdXEDwph8nQFaj8qEFry2cxoUveZCkPpNxQp4KhQwxuy4R7jXDMMsKkgW2yauC2dHbWYnr2Ee",
|
|
"1ef4e492",
|
|
"m/49h/0h/0h",
|
|
"sh-wpkh",
|
|
None,
|
|
"zpub6qk8ok1ouvwM1NkumKnsteGf1F9UUNshFdFdXEDwph8nQFaj8qEFry2cxoUveZCkPpNxQp4KhQwxuy4R7jXDMMsKkgW2yauC2dHbWYnr2Ee",
|
|
)
|
|
|
|
key4 = Key(
|
|
"[6ea15da6/49h/0h/0h]xpub6BtcNhqbaFaoC3oEfKky3Sm22pF48U2jmAf78cB3wdAkkGyAgmsVrgyt1ooSt3bHWgzsdUQh2pTJ867yTeUAMmFDKNSBp8J7WPmp7Df7zjv",
|
|
"6ea15da6",
|
|
"m/49h/0h/0h",
|
|
"sh-wpkh",
|
|
None,
|
|
"xpub6BtcNhqbaFaoC3oEfKky3Sm22pF48U2jmAf78cB3wdAkkGyAgmsVrgyt1ooSt3bHWgzsdUQh2pTJ867yTeUAMmFDKNSBp8J7WPmp7Df7zjv",
|
|
)
|
|
|
|
key5 = Key(
|
|
"[6ea15da6/49h/0h/0h]xpub6BtcNhqbaFaoG3xcuncx9xzL3X38FuWXdcdvsdG5Q99Cb4EgeVYPEYaVpX28he6472gEsCokg8v9oMVRTrZNe5LHtGSPcC5ofehYkhD1Kxy",
|
|
"6ea15da6",
|
|
"m/49h/0h/1h",
|
|
"sh-wpkh",
|
|
None, # slightly different ypub than key4
|
|
"xpub6BtcNhqbaFaoG3xcuncx9xzL3X38FuWXdcdvsdG5Q99Cb4EgeVYPEYaVpX28he6472gEsCokg8v9oMVRTrZNe5LHtGSPcC5ofehYkhD1Kxy",
|
|
)
|
|
|
|
# Case 1: Identical keys
|
|
keys = [key1, key1]
|
|
with pytest.raises(SpecterError):
|
|
wm._check_duplicate_keys(keys)
|
|
# Case 2: different keys
|
|
# key2 and 3 are different as they don't have the same xpub. See #1500 for discussion
|
|
keys = [key1, key2, key3] # key2 xpub is the same than key3 zpub
|
|
with pytest.raises(SpecterError):
|
|
wm._check_duplicate_keys(keys)
|
|
|
|
keys = [key4, key5]
|
|
wm._check_duplicate_keys(keys)
|
|
|
|
|
|
def test_wallet_sortedmulti(
|
|
bitcoin_regtest, devices_filled_data_folder, device_manager
|
|
):
|
|
wm = WalletManager(
|
|
devices_filled_data_folder,
|
|
bitcoin_regtest.get_rpc(),
|
|
"regtest",
|
|
device_manager,
|
|
)
|
|
device = device_manager.get_by_alias("trezor")
|
|
second_device = device_manager.get_by_alias("specter")
|
|
for i in range(2):
|
|
if i == 0:
|
|
multisig_wallet = wm.create_wallet(
|
|
"a_multisig_test_wallet",
|
|
1,
|
|
"wsh",
|
|
[device.keys[7], second_device.keys[0]],
|
|
[device, second_device],
|
|
)
|
|
else:
|
|
multisig_wallet = wm.create_wallet(
|
|
"a_multisig_test_wallet",
|
|
1,
|
|
"wsh",
|
|
[second_device.keys[0], device.keys[7]],
|
|
[second_device, device],
|
|
)
|
|
|
|
address = multisig_wallet.address
|
|
address_info = multisig_wallet.rpc.getaddressinfo(address)
|
|
assert address_info["pubkeys"][0] < address_info["pubkeys"][1]
|
|
|
|
another_address = multisig_wallet.getnewaddress()
|
|
another_address_info = multisig_wallet.rpc.getaddressinfo(another_address)
|
|
assert another_address_info["pubkeys"][0] < another_address_info["pubkeys"][1]
|
|
|
|
third_address = multisig_wallet.get_address(30)
|
|
third_address_info = multisig_wallet.rpc.getaddressinfo(third_address)
|
|
assert third_address_info["pubkeys"][0] < third_address_info["pubkeys"][1]
|
|
|
|
change_address = multisig_wallet.change_address
|
|
change_address_info = multisig_wallet.rpc.getaddressinfo(change_address)
|
|
assert change_address_info["pubkeys"][0] < change_address_info["pubkeys"][1]
|
|
|
|
another_change_address = multisig_wallet.get_address(30, change=True)
|
|
another_change_address_info = multisig_wallet.rpc.getaddressinfo(
|
|
another_change_address
|
|
)
|
|
assert (
|
|
another_change_address_info["pubkeys"][0]
|
|
< another_change_address_info["pubkeys"][1]
|
|
)
|
|
|
|
|
|
def test_wallet_labeling(bitcoin_regtest, devices_filled_data_folder, device_manager):
|
|
wm = WalletManager(
|
|
devices_filled_data_folder,
|
|
bitcoin_regtest.get_rpc(),
|
|
"regtest",
|
|
device_manager,
|
|
)
|
|
# A wallet-creation needs a device
|
|
device = device_manager.get_by_alias("specter")
|
|
key = Key.from_json(
|
|
{
|
|
"derivation": "m/48h/1h/0h/2h",
|
|
"original": "Vpub5n9kKePTPPGtw3RddeJWJe29epEyBBcoHbbPi5HhpoG2kTVsSCUzsad33RJUt3LktEUUPPofcZczuudnwR7ZgkAkT6N2K2Z7wdyjYrVAkXM",
|
|
"fingerprint": "08686ac6",
|
|
"type": "wsh",
|
|
"xpub": "tpubDFHpKypXq4kwUrqLotPs6fCic5bFqTRGMBaTi9s5YwwGymE8FLGwB2kDXALxqvNwFxB1dLWYBmmeFVjmUSdt2AsaQuPmkyPLBKRZW8BGCiL",
|
|
}
|
|
)
|
|
wallet = wm.create_wallet("a_second_test_wallet", 1, "wpkh", [key], [device])
|
|
|
|
first_address = wallet.address
|
|
assert wallet.getlabel(first_address) == "Address #0"
|
|
wallet.setlabel(first_address, "Random label")
|
|
assert wallet.getlabel(first_address) == "Random label"
|
|
|
|
second_address = wallet.getnewaddress()
|
|
wallet.setlabel(second_address, "")
|
|
bitcoin_regtest.testcoin_faucet(second_address, amount=0.6)
|
|
|
|
third_address = wallet.getnewaddress()
|
|
wallet.setlabel(third_address, "")
|
|
bitcoin_regtest.testcoin_faucet(third_address, amount=0.4)
|
|
|
|
assert sorted(wallet.addresses) == sorted(
|
|
[first_address, second_address, third_address]
|
|
)
|
|
|
|
# Make 20 UTXO
|
|
for i in range(0, 20):
|
|
_address = wallet.getnewaddress()
|
|
bitcoin_regtest.testcoin_faucet(_address, amount=0.1)
|
|
|
|
# update utxo
|
|
wallet.getdata()
|
|
assert len(wallet.full_utxo) == 22
|
|
|
|
# An entry in full_utxo looks something like this:
|
|
# { 'txid': 'fab823558781745179916b4bfdfd65b382bfc0e70e85188f1b9538604202f537',
|
|
# 'vout': 0, 'address': 'bcrt1qmlrraffw0evkjy2yrxmt263ksgfgv2gqhcddrt',
|
|
# 'label': 'Random label', 'scriptPubKey': '0014dfc63ea52e7e5969114419b6b56a368212862900',
|
|
# 'amount': 50.0, 'confirmations': 101, 'spendable': False, 'solvable': True,
|
|
# 'desc': "wpkh([08686ac6/48'/1'/0'/2'/0/0]02fa445808af849209038f422a22e335754fa07a2ece42fc483660606dcda3e0e9)#8q60z40m",
|
|
# 'safe': True, 'time': 1637091575, 'category': 'generate', 'locked': False
|
|
# }
|
|
|
|
|
|
def test_wallet_change_addresses(
|
|
bitcoin_regtest, devices_filled_data_folder, device_manager
|
|
):
|
|
wm = WalletManager(
|
|
devices_filled_data_folder,
|
|
bitcoin_regtest.get_rpc(),
|
|
"regtest",
|
|
device_manager,
|
|
)
|
|
# A wallet-creation needs a device
|
|
device = device_manager.get_by_alias("specter")
|
|
key = Key.from_json(
|
|
{
|
|
"derivation": "m/84h/1h/0h",
|
|
"original": "vpub5ZSem3mLXiSJzgDX6pJb2N9L6sJ8m6ejaksLPLSuB53LBzCi2mMsBg19eEUSDkHtyYp75GATjLgt5p3S43WjaVCXAWU9q9H5GhkwJBrMiAb",
|
|
"fingerprint": "08686ac6",
|
|
"type": "wpkh",
|
|
"xpub": "tpubDDUotcvrYMUiy4ncDirveTfhmvggdj8nxcW5JgHpGzYz3UVscJY5aEzFvgUPk4YyajadBnsTBmE2YZmAtJC14Q21xncJgVaHQ7UdqMRVRbU",
|
|
}
|
|
)
|
|
wallet: wallet = wm.create_wallet("a_third_test_wallet", 1, "wpkh", [key], [device])
|
|
|
|
address = wallet.address
|
|
change_address = wallet.change_address
|
|
assert wallet.addresses == [address]
|
|
assert wallet.change_addresses == [change_address]
|
|
bitcoin_regtest.testcoin_faucet(change_address, amount=0.1)
|
|
wallet.update_balance()
|
|
assert wallet.amount_total == 0.1
|
|
|
|
# new change address should be genrated automatically after receiving
|
|
# assert wallet.change_addresses == [change_address, wallet.change_address]
|
|
# This will not work here since Bitcoin Core doesn't count mining rewards in `getreceivedbyaddress`
|
|
# See: https://github.com/bitcoin/bitcoin/issues/14654
|
|
|
|
|
|
def test_singlesig_wallet_backup_and_restore(caplog, specter_regtest_configured, node):
|
|
"""
|
|
Single-sig wallets should be able to be backed up and re-imported with or without
|
|
the "devices" attr in the json backup.
|
|
"""
|
|
caplog.set_level(logging.INFO)
|
|
|
|
device_manager = specter_regtest_configured.device_manager
|
|
wallet_manager = specter_regtest_configured.wallet_manager
|
|
|
|
device = device_manager.get_by_alias("trezor")
|
|
device_type = device.device_type
|
|
|
|
# Get the 'wkph' testnet key
|
|
for key in device.keys:
|
|
if key.key_type == "wpkh" and key.xpub.startswith("tpub"):
|
|
break
|
|
|
|
# create a wallet
|
|
wallet = wallet_manager.create_wallet(
|
|
name="my_singlesig_test_wallet",
|
|
sigs_required=1,
|
|
key_type=key.key_type,
|
|
keys=[key],
|
|
devices=[device],
|
|
)
|
|
|
|
# Wallet was prefunded in specter_regtest_configured fixture, not sure, though, why the amount is 23 and not 20 ...
|
|
amount = wallet.amount_total
|
|
|
|
# Save the json backup
|
|
wallet_backup = json.loads(wallet.account_map)
|
|
assert "devices" in wallet_backup
|
|
|
|
# Clear everything out as if we've never seen this wallet or device before
|
|
wallet_manager.delete_wallet(wallet, node)
|
|
device_manager.remove_device(device, wallet_manager=wallet_manager)
|
|
assert wallet.name not in wallet_manager.wallets_names
|
|
assert device.name not in device_manager.devices_names
|
|
|
|
# Parse the backed up wallet (code adapted from the new_wallet endpoint)
|
|
(
|
|
wallet_name,
|
|
recv_descriptor,
|
|
cosigners_types,
|
|
) = WalletImporter.parse_wallet_data_import(wallet_backup)
|
|
|
|
descriptor = Descriptor.parse(
|
|
AddChecksum(recv_descriptor.split("#")[0]),
|
|
testnet=is_testnet(specter_regtest_configured.chain),
|
|
)
|
|
|
|
(
|
|
keys,
|
|
cosigners,
|
|
unknown_cosigners,
|
|
unknown_cosigners_types,
|
|
) = descriptor.parse_signers(device_manager.devices, cosigners_types)
|
|
|
|
device_name = cosigners_types[0]["label"]
|
|
assert device_name == "Trezor"
|
|
assert unknown_cosigners_types[0] == device_type
|
|
|
|
# Re-create the device
|
|
new_device = device_manager.add_device(
|
|
name=device_name,
|
|
device_type=unknown_cosigners_types[0],
|
|
keys=[unknown_cosigners[0][0]],
|
|
)
|
|
|
|
keys.append(unknown_cosigners[0][0])
|
|
cosigners.append(new_device)
|
|
|
|
wallet = wallet_manager.create_wallet(
|
|
name=wallet_name,
|
|
sigs_required=descriptor.multisig_M,
|
|
key_type=descriptor.address_type,
|
|
keys=keys,
|
|
devices=cosigners,
|
|
)
|
|
|
|
# Sync the new wallet in bitcoincore to its existing utxos.
|
|
wallet.rpc.rescanblockchain(0)
|
|
|
|
# We restored the wallet's utxos
|
|
wallet.update_balance()
|
|
assert wallet.amount_total == amount
|
|
|
|
# Now do it again, but without the newer "devices" attr
|
|
del wallet_backup["devices"]
|
|
|
|
# Clear everything out as if we've never seen this wallet or device before
|
|
wallet_manager.delete_wallet(wallet, node)
|
|
device_manager.remove_device(device, wallet_manager=wallet_manager)
|
|
assert wallet.name not in wallet_manager.wallets_names
|
|
assert device.name not in device_manager.devices_names
|
|
|
|
# Parse the backed up wallet (code adapted from the new_wallet endpoint)
|
|
(
|
|
wallet_name,
|
|
recv_descriptor,
|
|
cosigners_types,
|
|
) = WalletImporter.parse_wallet_data_import(wallet_backup)
|
|
|
|
descriptor = Descriptor.parse(
|
|
AddChecksum(recv_descriptor.split("#")[0]),
|
|
testnet=is_testnet(specter_regtest_configured.chain),
|
|
)
|
|
|
|
(
|
|
keys,
|
|
cosigners,
|
|
unknown_cosigners,
|
|
unknown_cosigners_types,
|
|
) = descriptor.parse_signers(device_manager.devices, cosigners_types)
|
|
|
|
assert len(cosigners_types) == 0
|
|
assert unknown_cosigners_types[0] == GenericDevice.device_type
|
|
|
|
# Re-create the device
|
|
new_device = device_manager.add_device(
|
|
name=device_name,
|
|
device_type=unknown_cosigners_types[0],
|
|
keys=[unknown_cosigners[0][0]],
|
|
)
|
|
|
|
keys.append(unknown_cosigners[0][0])
|
|
cosigners.append(new_device)
|
|
|
|
wallet = wallet_manager.create_wallet(
|
|
name=wallet_name,
|
|
sigs_required=descriptor.multisig_M,
|
|
key_type=descriptor.address_type,
|
|
keys=keys,
|
|
devices=cosigners,
|
|
)
|
|
|
|
# Sync the new wallet in bitcoincore to its existing utxos
|
|
wallet.rpc.rescanblockchain(0)
|
|
|
|
# We restored the wallet's utxos
|
|
wallet.update_balance()
|
|
assert wallet.amount_total == amount
|
|
|
|
|
|
def test_multisig_wallet_backup_and_restore(
|
|
bitcoin_regtest, caplog, specter_regtest_configured, node
|
|
):
|
|
"""
|
|
Multisig wallets should be able to be backed up and re-imported
|
|
with or without the "devices" attr in the json backup.
|
|
"""
|
|
caplog.set_level(logging.INFO)
|
|
|
|
device_manager = specter_regtest_configured.device_manager
|
|
wallet_manager = specter_regtest_configured.wallet_manager
|
|
|
|
device = device_manager.get_by_alias("trezor")
|
|
device_type = device.device_type
|
|
|
|
# Get the multisig 'wsh' testnet key
|
|
for key in device.keys:
|
|
if key.key_type == "wsh" and key.xpub.startswith("tpub"):
|
|
break
|
|
|
|
# Create a pair of hot wallet signers
|
|
hot_wallet_1_device = device_manager.add_device(
|
|
name="hot_key_1", device_type=BitcoinCore.device_type, keys=[]
|
|
)
|
|
hot_wallet_1_device.setup_device(file_password=None, wallet_manager=wallet_manager)
|
|
hot_wallet_1_device.add_hot_wallet_keys(
|
|
mnemonic=generate_mnemonic(strength=128),
|
|
passphrase="",
|
|
paths=["m/48h/1h/0h/2h"],
|
|
file_password=None,
|
|
wallet_manager=wallet_manager,
|
|
testnet=True,
|
|
keys_range=[0, 1000],
|
|
keys_purposes=[],
|
|
)
|
|
hot_wallet_2_device = device_manager.add_device(
|
|
name="hot_key_2", device_type=BitcoinCore.device_type, keys=[]
|
|
)
|
|
hot_wallet_2_device.setup_device(file_password=None, wallet_manager=wallet_manager)
|
|
hot_wallet_2_device.add_hot_wallet_keys(
|
|
mnemonic=generate_mnemonic(strength=128),
|
|
passphrase="",
|
|
paths=["m/48h/1h/0h/2h"],
|
|
file_password=None,
|
|
wallet_manager=wallet_manager,
|
|
testnet=True,
|
|
keys_range=[0, 1000],
|
|
keys_purposes=[],
|
|
)
|
|
|
|
# create the multisig wallet
|
|
wallet = wallet_manager.create_wallet(
|
|
name="my_multisig_test_wallet",
|
|
sigs_required=2,
|
|
key_type=key.key_type,
|
|
keys=[key, hot_wallet_1_device.keys[0], hot_wallet_2_device.keys[0]],
|
|
devices=[device, hot_wallet_1_device, hot_wallet_2_device],
|
|
)
|
|
|
|
# Wallet is unfunded
|
|
address = wallet.getnewaddress()
|
|
bitcoin_regtest.testcoin_faucet(address, amount=3.3)
|
|
wallet.update_balance()
|
|
assert wallet.amount_total == 3.3
|
|
|
|
# Save the json backup
|
|
wallet_backup = json.loads(wallet.account_map.replace("\\\\", "").replace("'", "h"))
|
|
assert "devices" in wallet_backup
|
|
|
|
# Clear everything out as if we've never seen this wallet or device before
|
|
wallet_manager.delete_wallet(wallet, node)
|
|
device_manager.remove_device(device, wallet_manager=wallet_manager)
|
|
assert wallet.name not in wallet_manager.wallets_names
|
|
assert device.name not in device_manager.devices_names
|
|
|
|
# Parse the backed up wallet (code adapted from the new_wallet endpoint)
|
|
(
|
|
wallet_name,
|
|
recv_descriptor,
|
|
cosigners_types,
|
|
) = WalletImporter.parse_wallet_data_import(wallet_backup)
|
|
|
|
descriptor = Descriptor.parse(
|
|
AddChecksum(recv_descriptor.split("#")[0]),
|
|
testnet=is_testnet(specter_regtest_configured.chain),
|
|
)
|
|
|
|
(
|
|
keys,
|
|
cosigners,
|
|
unknown_cosigners,
|
|
unknown_cosigners_types,
|
|
) = descriptor.parse_signers(device_manager.devices, cosigners_types)
|
|
|
|
assert cosigners_types[0]["label"] == "Trezor"
|
|
assert cosigners_types[0]["type"] == device_type
|
|
|
|
assert cosigners_types[1]["label"] == "hot_key_1"
|
|
assert cosigners_types[1]["type"] == BitcoinCore.device_type
|
|
|
|
assert cosigners_types[2]["label"] == "hot_key_2"
|
|
assert cosigners_types[2]["type"] == BitcoinCore.device_type
|
|
|
|
# Re-create the Trezor device
|
|
new_device = device_manager.add_device(
|
|
name=unknown_cosigners[0][1],
|
|
device_type=unknown_cosigners_types[0],
|
|
keys=[unknown_cosigners[0][0]],
|
|
)
|
|
keys.append(unknown_cosigners[0][0])
|
|
cosigners.append(new_device)
|
|
|
|
wallet = wallet_manager.create_wallet(
|
|
name=wallet_name,
|
|
sigs_required=descriptor.multisig_M,
|
|
key_type=descriptor.address_type,
|
|
keys=keys,
|
|
devices=cosigners,
|
|
)
|
|
|
|
# Sync the new wallet in bitcoincore to its existing utxos
|
|
wallet.rpc.rescanblockchain(0)
|
|
|
|
# We restored the wallet's utxos
|
|
assert wallet.amount_total == 3.3
|
|
|
|
# Now do it again, but without the newer "devices" attr
|
|
del wallet_backup["devices"]
|
|
|
|
# Clear everything out as if we've never seen this wallet or device before
|
|
wallet_manager.delete_wallet(wallet, node)
|
|
for device_names in device_manager.devices:
|
|
device = device_manager.devices[device_names]
|
|
device_manager.remove_device(device, wallet_manager=wallet_manager)
|
|
assert wallet.name not in wallet_manager.wallets_names
|
|
assert device.name not in device_manager.devices_names
|
|
|
|
# Parse the backed up wallet (code adapted from the new_wallet endpoint)
|
|
(
|
|
wallet_name,
|
|
recv_descriptor,
|
|
cosigners_types,
|
|
) = WalletImporter.parse_wallet_data_import(wallet_backup)
|
|
|
|
descriptor = Descriptor.parse(
|
|
AddChecksum(recv_descriptor.split("#")[0]),
|
|
testnet=is_testnet(specter_regtest_configured.chain),
|
|
)
|
|
|
|
(
|
|
keys,
|
|
cosigners,
|
|
unknown_cosigners,
|
|
unknown_cosigners_types,
|
|
) = descriptor.parse_signers(device_manager.devices, cosigners_types)
|
|
|
|
# Now we don't know any of the cosigners' types
|
|
assert len(cosigners_types) == 0
|
|
assert unknown_cosigners_types[0] == GenericDevice.device_type
|
|
|
|
# Re-create all three devices
|
|
for i, (unknown_cosigner_key, label) in enumerate(unknown_cosigners):
|
|
# 'label' will be unknown
|
|
assert label is None
|
|
new_device = device_manager.add_device(
|
|
name=f"{wallet_name} signer {i + 1}",
|
|
device_type=unknown_cosigners_types[i],
|
|
keys=[unknown_cosigner_key],
|
|
)
|
|
keys.append(unknown_cosigner_key)
|
|
cosigners.append(new_device)
|
|
|
|
wallet = wallet_manager.create_wallet(
|
|
name=wallet_name,
|
|
sigs_required=descriptor.multisig_M,
|
|
key_type=descriptor.address_type,
|
|
keys=keys,
|
|
devices=cosigners,
|
|
)
|
|
|
|
# Sync the new wallet in bitcoincore to its existing utxos
|
|
wallet.rpc.rescanblockchain(0)
|
|
|
|
# We restored the wallet's utxos
|
|
assert wallet.amount_total == 3.3
|
|
|
|
|
|
def test_threading(specter_regtest_configured_with_threading):
|
|
assert (
|
|
specter_regtest_configured_with_threading.config["testing"][
|
|
"allow_threading_for_testing"
|
|
]
|
|
== True
|
|
)
|
|
device = specter_regtest_configured_with_threading.device_manager.get_by_alias(
|
|
"trezor"
|
|
)
|
|
wm = specter_regtest_configured_with_threading.wallet_manager
|
|
wallet = wm.create_wallet("test_wallet", 1, "wpkh", [device.keys[5]], [device])
|
|
assert wm.wallets_names == ["test_wallet"]
|
|
assert wm.data_folder.endswith("wallets")
|