Chore: Refactor device_types and disable change to hotwallet for Spectrum_Node (#2048)

* adding fullpath fixes spectrum issue 13

* Add Spectrum node delte capability

* refactor and enable get_subclass_for_clazz for all classes

* get rid of DeviceTypes references part 1

* get rid of DeviceTypes references part 2

* add check for device-type when changing device_type

* added return type hint

* remove fatal logging

* Change flash message

* Disable setting unsupported device typ in UI

Co-authored-by: Manolis Mandrapilias <70536101+moneymanolis@users.noreply.github.com>
Co-authored-by: moneymanolis <moneymanolis@protonmail.com>
This commit is contained in:
k9ert 2023-01-26 15:05:45 +01:00 committed by GitHub
parent ff94b06511
commit d367e93df6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
27 changed files with 114 additions and 104 deletions

View file

@ -1,4 +1,7 @@
import json
from typing import Type
from cryptoadvance.specter.util.reflection import get_subclasses_for_clazz
from .key import Key
from .persistence import read_json_file, write_json_file
import logging
@ -59,6 +62,13 @@ class Device:
return f"{module_array[2]}_endpoint.static"
return "static"
@classmethod
def get_device_class_by_device_type_string(cls, device_type: str) -> Type:
for clazz in get_subclasses_for_clazz(Device):
if clazz.device_type == device_type:
return clazz
return None
def create_psbts(self, base64_psbt, wallet):
"""
Overwrite this method for a device.

View file

@ -1,4 +1,3 @@
from .device_types import DeviceTypes
from .coldcard import ColdCard
from .trezor import Trezor
from .ledger import Ledger

View file

@ -1,9 +1,8 @@
from . import DeviceTypes
from .hwi_device import HWIDevice
class BitBox02(HWIDevice):
device_type = DeviceTypes.BITBOX02
device_type = "bitbox02"
name = "BitBox02"
icon = "img/devices/bitbox02_icon.svg"
supports_hwi_multisig_display_address = True

View file

@ -15,13 +15,12 @@ from ..util.base58 import decode_base58, encode_base58_checksum
from ..util.descriptor import AddChecksum
from ..util.xpub import convert_xpub_prefix, get_xpub_fingerprint
from ..util.mnemonic import mnemonic_to_root
from . import DeviceTypes
logger = logging.getLogger(__name__)
class BitcoinCore(Device):
device_type = DeviceTypes.BITCOINCORE
device_type = "bitcoincore"
name = "Bitcoin Core (hot wallet)"
icon = "img/devices/bitcoincore_icon.svg"
@ -271,7 +270,7 @@ class BitcoinCoreWatchOnly(BitcoinCore):
It can be converted back to a device of Type BitcoinCore by providing the 12 words again.
"""
device_type = DeviceTypes.BITCOINCORE_WATCHONLY
device_type = "bitcoincore_watchonly"
name = "Bitcoin Core (watch only)"
hot_wallet = False
@ -306,6 +305,6 @@ class BitcoinCoreWatchOnly(BitcoinCore):
)
# Change type (also triggers write to file)
self.set_type(DeviceTypes.BITCOINCORE)
self.set_type(BitcoinCore.device_type)
# After update this device will be available as a BitcoinCore (hot) instance
self.manager.update(comment="via BitcoinCoreWatchOnly")

View file

@ -1,7 +1,6 @@
import hashlib
# from ..device import Device
from . import DeviceTypes
from .coldcard import ColdCard
from hwilib.psbt import PSBT
from binascii import a2b_base64
@ -11,7 +10,7 @@ from ..helpers import to_ascii20
class Cobo(ColdCard):
device_type = DeviceTypes.COBO
device_type = "cobo"
name = "Cobo Vault"
icon = "img/devices/cobo_icon.svg"

View file

@ -1,27 +0,0 @@
""" A list of static device_type strings. As Devices can, in the meantime, also be created
in extensions, this file should be limited to device_type_strings which ARE referenced
in core-code.
ToDo: Remove the strings which are only referenced in the corresponding-type.
There is no point in doing that.
"""
class DeviceTypes:
BITCOINCORE = "bitcoincore"
BITCOINCORE_WATCHONLY = "bitcoincore_watchonly"
ELEMENTSCORE = "elementscore"
# All those ones are not treated any different in core-code and therefore should no
# longer be referenced here:
BITBOX02 = "bitbox02"
COBO = "cobo"
COLDCARD = "coldcard"
GENERICDEVICE = "other"
JADE = "jade"
KEEPKEY = "keepkey"
LEDGER = "ledger"
SEEDSIGNER = "seedsigner"
SPECTERDIY = "specter"
TREZOR = "trezor"
KEYSTONE = "keystone"
PASSPORT = "passport"

View file

@ -3,12 +3,11 @@ from embit import bip39
from embit.liquid.slip77 import master_blinding_from_seed
from ..helpers import is_liquid
from . import DeviceTypes
from .bitcoin_core import BitcoinCore
class ElementsCore(BitcoinCore):
device_type = DeviceTypes.ELEMENTSCORE
device_type = "elementscore"
name = "Elements Core (hot wallet)"
icon = "img/devices/elementscore_icon.svg"

View file

@ -1,9 +1,8 @@
from . import DeviceTypes
from ..device import Device
class GenericDevice(Device):
device_type = DeviceTypes.GENERICDEVICE
device_type = "other"
name = "Other"
sd_card_support = True

View file

@ -1,4 +1,3 @@
from . import DeviceTypes
from .hwi_device import HWIDevice
from .hwi.jade import JadeClient
from .hwi.jade import enumerate as jade_enumerate
@ -6,7 +5,7 @@ from ..helpers import is_liquid
class Jade(HWIDevice):
device_type = DeviceTypes.JADE
device_type = "jade"
name = "Jade"
icon = "img/devices/jade_icon.svg"

View file

@ -1,10 +1,9 @@
from . import DeviceTypes
from .hwi_device import HWIDevice
from hwilib.devices.keepkey import KeepkeyClient
class Keepkey(HWIDevice):
device_type = DeviceTypes.KEEPKEY
device_type = "keepkey"
name = "KeepKey"
icon = "img/devices/keepkey_icon.svg"

View file

@ -1,4 +1,3 @@
from . import DeviceTypes
from .cobo import Cobo
from ..helpers import to_ascii20
from ..util.xpub import get_xpub_fingerprint
@ -6,7 +5,7 @@ from binascii import b2a_base64
class Keystone(Cobo):
device_type = DeviceTypes.KEYSTONE
device_type = "keystone"
name = "Keystone"
icon = "img/devices/keystone_icon.svg"

View file

@ -1,9 +1,8 @@
from . import DeviceTypes
from .hwi_device import HWIDevice
class Ledger(HWIDevice):
device_type = DeviceTypes.LEDGER
device_type = "ledger"
name = "Ledger"
icon = "img/devices/ledger_icon.svg"

View file

@ -1,8 +1,7 @@
from . import DeviceTypes
from .keystone import Keystone
class Passport(Keystone):
device_type = DeviceTypes.PASSPORT
device_type = "passport"
name = "Passport"
icon = "img/devices/passport_icon.svg"

View file

@ -1,9 +1,8 @@
from . import DeviceTypes
from ..device import Device
class SeedSignerDevice(Device):
device_type = DeviceTypes.SEEDSIGNER
device_type = "seedsigner"
name = "SeedSigner"
icon = "img/devices/seedsigner_icon.svg"

View file

@ -1,5 +1,4 @@
import hashlib
from . import DeviceTypes
from .sd_card_device import SDCardDevice
from .hwi.specter_diy import enumerate as specter_enumerate, SpecterClient
from ..helpers import to_ascii20
@ -60,7 +59,7 @@ def fill_external_wallet_derivations(psbt, wallet):
class Specter(SDCardDevice):
device_type = DeviceTypes.SPECTERDIY
device_type = "specter"
name = "Specter-DIY"
icon = "img/devices/specter_icon.svg"

View file

@ -1,10 +1,9 @@
from . import DeviceTypes
from .hwi_device import HWIDevice
from hwilib.devices.trezor import TrezorClient
class Trezor(HWIDevice):
device_type = DeviceTypes.TREZOR
device_type = "trezor"
name = "Trezor"
icon = "img/devices/trezor_icon.svg"

View file

@ -3,16 +3,21 @@ import json
import logging
import random
import re
from cryptoadvance.specter.device import Device
from cryptoadvance.specter.devices.device_types import DeviceTypes
from flask import Blueprint, Flask
from flask import current_app as app
from flask import jsonify, redirect, render_template, request, url_for
from flask_babel import lazy_gettext as _
from flask_login import current_user, login_required
from mnemonic import Mnemonic
from cryptoadvance.specter.devices.elements_core import ElementsCore
from cryptoadvance.specter.managers.node_manager import NodeManager
from cryptoadvance.specter.node import Node
from ..devices.bitcoin_core import BitcoinCore
from cryptoadvance.specterext.electrum.controller import specter
from ..devices.bitcoin_core import BitcoinCore, BitcoinCoreWatchOnly
from ..helpers import is_testnet
from ..key import Key
from ..managers.device_manager import get_device_class
@ -92,15 +97,15 @@ def new_device_keys(device_type):
keys.append(Key.parse_xpub(request.form["master_pub_key"]))
if not keys and not err:
if device_type in [
DeviceTypes.BITCOINCORE,
DeviceTypes.ELEMENTSCORE,
DeviceTypes.BITCOINCORE_WATCHONLY,
BitcoinCore.device_type,
ElementsCore.device_type,
BitcoinCoreWatchOnly.device_type,
]:
if not paths:
err = _("No paths were specified, please provide at least one.")
if err is None:
if existing_device:
if device_type == DeviceTypes.BITCOINCORE_WATCHONLY:
if device_type == BitcoinCoreWatchOnly.device_type:
device.setup_device(
file_password, app.specter.wallet_manager
)
@ -404,7 +409,7 @@ def new_device_manual():
def device(device_alias):
err = None
try:
device = app.specter.device_manager.get_by_alias(device_alias)
device: Device = app.specter.device_manager.get_by_alias(device_alias)
except:
return render_template(
"base.jinja", error=_("Device not found"), specter=app.specter, rand=rand
@ -490,7 +495,16 @@ def device(device_alias):
)
elif action == "settype":
device_type = request.form["device_type"]
device.set_type(device_type)
if app.specter.node_manager.active_node.is_device_supported(
Device.get_device_class_by_device_type_string(device_type)
):
device.set_type(device_type)
# Should actually not be possible to end up here (UI doesn't let you select this) but just in case
else:
flash(
f"The device type {Device.get_device_class_by_device_type_string(device_type).name} is not supported by your connection.",
"error",
)
device = copy.deepcopy(device)
def sort_accounts(k):

View file

@ -15,7 +15,7 @@ from sys import exit
from urllib.parse import urlparse
import requests
from cryptoadvance.specter.devices.device_types import DeviceTypes
from cryptoadvance.specter.devices.bitcoin_core import BitcoinCore, BitcoinCoreWatchOnly
from cryptoadvance.specter.services.service_encrypted_storage import (
ServiceEncryptedStorageManager,
ServiceUnencryptedStorageManager,
@ -729,8 +729,8 @@ class Specter:
data.compress_type = zipfile.ZIP_DEFLATED
device = device.json
# Exporting the bitcoincore hot wallet as watchonly
if device["type"] == DeviceTypes.BITCOINCORE:
device["type"] = DeviceTypes.BITCOINCORE_WATCHONLY
if device["type"] == BitcoinCore.device_type:
device["type"] = BitcoinCoreWatchOnly.device_type
zf.writestr(
"devices/{}.json".format(device["alias"]), json.dumps(device)
)

View file

@ -5,8 +5,10 @@
<select name="device_type" id="device_type">
<option value="" selected>{{ _("Select type") }}</option>
{% for cls in specter.device_manager.supported_devices %}
{% if cls.device_type != "bitcoincore" or (device and device.device_type != "bitcoincore_watchonly") %}
<option value="{{cls.device_type}}">{{cls.name}}</option>
{% if specter.node_manager.active_node.is_device_supported(cls) and cls.device_type != 'elementscore' %}
<option value="{{cls.device_type}}">{{ cls.name }}</option>
{% else %}
<option value="{{cls.device_type}}" disabled>{{ cls.name }}</option>
{% endif %}
{% endfor %}
</select>

View file

@ -1,6 +1,6 @@
import logging
from importlib import import_module
from inspect import isclass
import inspect
import os
from pathlib import Path
import pkgutil
@ -106,7 +106,7 @@ def get_classlist_of_type_clazz_from_modulelist(clazz, modulelist):
logger.debug(f"Imported {fq_module_name}")
for attribute_name in dir(module):
attribute = getattr(module, attribute_name)
if isclass(attribute):
if inspect.isclass(attribute):
if (
issubclass(attribute, clazz)
# This works for 1 level inheritance within one module
@ -123,7 +123,6 @@ def get_classlist_of_type_clazz_from_modulelist(clazz, modulelist):
def get_subclasses_for_clazz_in_cwd(clazz, cwd=".") -> List[type]:
"""Returns all subclasses of class clazz located in the CWD if the cwd
is not a specter-desktop dev-env-kind-of-dir or contains any .py-file
So
"""
package_dirs = []
# security first! No dynamic loading in app-images
@ -158,8 +157,10 @@ def get_subclasses_for_clazz_in_cwd(clazz, cwd=".") -> List[type]:
return get_subclasses_for_clazz(clazz, package_dirs)
def get_subclasses_for_clazz(clazz, package_dirs: List[str] = None):
"""Returns all subclasses of class clazz located in the CWD
def get_subclasses_for_clazz(
clazz, package_dirs: List[str] = None, package: str = None
):
"""Returns all subclasses of class clazz searching in specific directories
potentially add additional_packagedirs which is usefull for
calculating pyinstaller hiddenimports
"""
@ -222,21 +223,13 @@ def get_subclasses_for_clazz(clazz, package_dirs: List[str] = None):
f"cryptoadvance.specter.util.migrations.{module_name}"
)
else:
try:
module = import_module(
f"{module_name}.{camelcase2snake_case(clazz.__name__)}"
)
logger.debug(
f"Imported {module_name}.{camelcase2snake_case(clazz.__name__)}"
)
except ModuleNotFoundError as e:
logger.debug(
f"No Service Impl found in {module_name}.service. Skipping!"
)
continue
# Hopefully all the classes we're searching for are imported otherwise not much get
# found
return get_subclasses(clazz)
for attribute_name in dir(module):
attribute = getattr(module, attribute_name)
if isclass(attribute):
if inspect.isclass(attribute):
if (
issubclass(attribute, clazz)
and not attribute.__name__ == clazz.__name__
@ -244,3 +237,17 @@ def get_subclasses_for_clazz(clazz, package_dirs: List[str] = None):
class_list.append(attribute)
logger.info(f" Found class {attribute.__name__}")
return class_list
def get_subclasses(cls):
"""Naive implementation of searching for subclasses. This will only return classes which has been
imported. If you also want to find classes which are not imported, you need to provide package-directories
or well known places as for specific classes as implemented in get_subclasses_for_clazz
Returns all subclasses of a specific call including sususub...classes
"""
subclasses = []
for subclass in cls.__subclasses__():
# if not subclass.__module__.startswith("test_"):
subclasses.append(subclass)
subclasses.extend(get_subclasses(subclass))
return subclasses

View file

@ -1,7 +1,6 @@
""" This Device is not a real device but just something to demonstrate the creation of devices in extensions
"""
from cryptoadvance.specter.devices.device_types import DeviceTypes
from cryptoadvance.specter.devices.hwi_device import HWIDevice

View file

@ -1,6 +1,5 @@
from binascii import a2b_base64
from typing import List
from cryptoadvance.specter.devices.device_types import DeviceTypes
from cryptoadvance.specter.util.base43 import b43_encode
from cryptoadvance.specter.device import Device

View file

@ -1,4 +1,6 @@
import logging
from cryptoadvance.specter.util.reflection import get_class
from cryptoadvance.specter.util.common import snake_case2camelcase
from cryptoadvance.specterext.spectrum.bridge_rpc import BridgeRPC
from cryptoadvance.specter.helpers import deep_update
from cryptoadvance.specter.node import AbstractNode
@ -198,8 +200,18 @@ class SpectrumNode(AbstractNode):
# If a device class is passed as argument, take that, otherwise derive the class from the instance
if device_class_or_device_instance.__class__ == type:
device_class = device_class_or_device_instance
elif device_class_or_device_instance.__class__ == str:
if device_class_or_device_instance.__contains__("."):
device_class = get_class(device_class)
else:
fqcn_device_class = (
"cryptoadvance.specter.devices."
+ snake_case2camelcase(device_class_or_device_instance)
)
device_class = get_class(fqcn_device_class)
else:
device_class = device_class_or_device_instance.__class__
logger.debug(f"Device_class = {device_class}")
if device_class == BitcoinCore:
return False
return True

View file

@ -1,5 +1,6 @@
import json, requests
from cryptoadvance.specter.devices import DeviceTypes
from cryptoadvance.specter.devices.ledger import Ledger
from cryptoadvance.specter.devices.trezor import Trezor
def test_malformed_parse_error(client):
@ -134,7 +135,7 @@ def test_call_not_connected_device(client):
"method": "prompt_pin",
"id": 1,
"params": {
"device_type": DeviceTypes.TREZOR,
"device_type": Trezor.device_type,
"path": "",
"passphrase": "",
"chain": "test",
@ -161,7 +162,7 @@ def test_call_prompt_pin_invalid_device(client):
"method": "prompt_pin",
"id": 1,
"params": {
"device_type": DeviceTypes.LEDGER,
"device_type": Ledger.device_type,
"path": "",
"passphrase": "",
"chain": "test",

View file

@ -69,6 +69,7 @@ def test_ServiceManager_get_service_x_dirs(caplog):
def test_ServiceManager_get_service_packages(caplog):
caplog.set_level(logging.DEBUG)
packages = ServiceManager.get_service_packages()
assert "cryptoadvance.specterext.electrum.service" in packages
assert "cryptoadvance.specterext.electrum.devices.electrum" in packages

View file

@ -5,13 +5,14 @@ import time
from unittest.mock import MagicMock
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.devices import DeviceTypes
from cryptoadvance.specter.managers.wallet_manager import WalletManager
from cryptoadvance.specter.specter_error import SpecterError
from cryptoadvance.specter.util.descriptor import AddChecksum, Descriptor
@ -493,7 +494,7 @@ def test_singlesig_wallet_backup_and_restore(caplog, specter_regtest_configured,
) = descriptor.parse_signers(device_manager.devices, cosigners_types)
assert len(cosigners_types) == 0
assert unknown_cosigners_types[0] == DeviceTypes.GENERICDEVICE
assert unknown_cosigners_types[0] == GenericDevice.device_type
# Re-create the device
new_device = device_manager.add_device(
@ -543,7 +544,7 @@ def test_multisig_wallet_backup_and_restore(
# Create a pair of hot wallet signers
hot_wallet_1_device = device_manager.add_device(
name="hot_key_1", device_type=DeviceTypes.BITCOINCORE, keys=[]
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(
@ -557,7 +558,7 @@ def test_multisig_wallet_backup_and_restore(
keys_purposes=[],
)
hot_wallet_2_device = device_manager.add_device(
name="hot_key_2", device_type=DeviceTypes.BITCOINCORE, keys=[]
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(
@ -619,10 +620,10 @@ def test_multisig_wallet_backup_and_restore(
assert cosigners_types[0]["type"] == device_type
assert cosigners_types[1]["label"] == "hot_key_1"
assert cosigners_types[1]["type"] == DeviceTypes.BITCOINCORE
assert cosigners_types[1]["type"] == BitcoinCore.device_type
assert cosigners_types[2]["label"] == "hot_key_2"
assert cosigners_types[2]["type"] == DeviceTypes.BITCOINCORE
assert cosigners_types[2]["type"] == BitcoinCore.device_type
# Re-create the Trezor device
new_device = device_manager.add_device(
@ -679,7 +680,7 @@ def test_multisig_wallet_backup_and_restore(
# Now we don't know any of the cosigners' types
assert len(cosigners_types) == 0
assert unknown_cosigners_types[0] == DeviceTypes.GENERICDEVICE
assert unknown_cosigners_types[0] == GenericDevice.device_type
# Re-create all three devices
for i, (unknown_cosigner_key, label) in enumerate(unknown_cosigners):

View file

@ -5,6 +5,7 @@ from typing import List
from cryptoadvance.specter.device import Device
from cryptoadvance.specter.devices.bitbox02 import BitBox02
from cryptoadvance.specter.specter_error import SpecterInternalException
from cryptoadvance.specter.util.migrations.migration_0001 import SpecterMigration_0001
from cryptoadvance.specter.util.reflection import (
get_class,
get_subclasses_for_clazz,
@ -19,6 +20,7 @@ from cryptoadvance.specter.util.migrations.migration_0000 import SpecterMigratio
from cryptoadvance.specter.services.service import Service
from cryptoadvance.specterext.devhelp.service import DevhelpService
from cryptoadvance.specterext.electrum.service import ElectrumService
from cryptoadvance.specterext.spectrum.service import SpectrumService
from cryptoadvance.specterext.swan.service import SwanService
@ -86,9 +88,12 @@ def test_get_subclasses_for_clazz_in_cwd(caplog):
def test_get_subclasses_for_class(caplog):
caplog.set_level(logging.DEBUG)
classlist = get_subclasses_for_clazz(SpecterMigration)
assert SpecterMigration_0000 in classlist
assert len(classlist) >= 3
classlist = get_subclasses_for_clazz(Service)
assert len(classlist) == 4 # Happy to remove that at some point
assert SwanService in classlist
assert ElectrumService in classlist
assert DevhelpService in classlist
assert len(classlist) >= 4
# checking naively for certain Services would be counterproductive if you import
# the class. THis needs to work without importing the class!
# But we can test like this:
assert "SwanService" in [cls.__name__ for cls in classlist]
classlist = get_subclasses_for_clazz(Device)
assert len(classlist) > 5