<commit_before>from binascii import hexlify, unhexlify
import traceback
import sys
from typing import NamedTuple, Any, Optional, Dict, Union, List, Tuple, TYPE_CHECKING

from electrum.util import bfh, bh2u, UserCancelled, UserFacingException
from electrum.bip32 import BIP32Node
from electrum import constants
from electrum.i18n import _
from electrum.transaction import Transaction, PartialTransaction, PartialTxInput, PartialTxOutput
from electrum.keystore import Hardware_KeyStore
from electrum.base_wizard import ScriptTypeNotSupported

from ..hw_wallet import HW_PluginBase
from ..hw_wallet.plugin import (is_any_tx_output_on_change_branch, trezor_validate_op_return_output_and_get_data,
                                get_xpubs_and_der_suffixes_from_txinout)

if TYPE_CHECKING:
    from .client import KeepKeyClient

# TREZOR initialization methods
TIM_NEW, TIM_RECOVER, TIM_MNEMONIC, TIM_PRIVKEY = range(0, 4)


class KeepKey_KeyStore(Hardware_KeyStore):
    hw_type = 'keepkey'
    device = 'KeepKey'

    plugin: 'KeepKeyPlugin'

    def get_client(self, force_pair=True):
        return self.plugin.get_client(self, force_pair)

    def decrypt_message(self, sequence, message, password):
        raise UserFacingException(_('Encryption and decryption are not implemented by {}').format(self.device))

    def sign_message(self, sequence, message, password):
        client = self.get_client()
        address_path = self.get_derivation_prefix() + "/%d/%d"%sequence
        address_n = client.expand_path(address_path)
        msg_sig = client.sign_message(self.plugin.get_coin_name(), address_n, message)
        return msg_sig.signature

    def sign_transaction(self, tx, password):
        if tx.is_complete():
            return
        # previous transactions used as inputs
        prev_tx = {}
        for txin in tx.inputs():
            tx_hash = txin.prevout.txid.hex()
            if txin.utxo is None and not Transaction.is_segwit_input(txin):
                raise UserFacingException(_('Missing previous tx for legacy input.'))
            prev_tx[tx_hash] = txin.utxo

        self.plugin.sign_transaction(self, tx, prev_tx)


class KeepKeyPlugin(HW_PluginBase):
    # Derived classes provide:
    #
    #  class-static variables: client_class, firmware_URL, handler_class,
    #     libraries_available, libraries_URL, minimum_firmware,
    #     wallet_class, ckd_public, types, HidTransport

    firmware_URL = 'https://www.keepkey.com'
    libraries_URL = 'https://github.com/keepkey/python-keepkey'
    minimum_firmware = (1, 0, 0)
    keystore_class = KeepKey_KeyStore
    SUPPORTED_XTYPES = ('standard', 'p2wpkh-p2sh', 'p2wpkh', 'p2wsh-p2sh', 'p2wsh')

    MAX_LABEL_LEN = 32

    def __init__(self, parent, config, name):
        HW_PluginBase.__init__(self, parent, config, name)

        try:
            from . import client
            import keepkeylib
            import keepkeylib.ckd_public
            import keepkeylib.transport_hid
            import keepkeylib.transport_webusb
            self.client_class = client.KeepKeyClient
            self.ckd_public = keepkeylib.ckd_public
            self.types = keepkeylib.client.types
            self.DEVICE_IDS = (keepkeylib.transport_hid.DEVICE_IDS +
                               keepkeylib.transport_webusb.DEVICE_IDS)
            self.device_manager().register_devices(self.DEVICE_IDS)
            self.libraries_available = True
        except ImportError:
            self.libraries_available = False

    def hid_transport(self, pair):
        from keepkeylib.transport_hid import HidTransport
        return HidTransport(pair)

    def webusb_transport(self, device):
        from keepkeylib.transport_webusb import WebUsbTransport
        for d in WebUsbTransport.enumerate():
            if device.id_.startswith(d.getSerialNumber()):
                return WebUsbTransport(d)
        return WebUsbTransport(device)

    def _try_hid(self, device):
        self.logger.info("Trying to connect over USB...")
        if device.interface_number == 1:
            pair = [None, device.path]
        else:
            pair = [device.path, None]

        try:
            return self.hid_transport(pair)
        except BaseException as e:
            # see fdb810ba622dc7dbe1259cbafb5b28e19d2ab114
            # raise
            self.logger.info(f"cannot connect at {device.path} {e}")
            return None

    def _try_webusb(self, device):
        self.logger.info("Trying to connect over WebUSB...")
        try:
            return self.webusb_transport(device)
        except BaseException as e:
            self.logger.info(f"cannot connect at {device.path} {e}")
            return None<commit_msg>keepkey: use libusb to enumerate devices instead of hid<commit_after>from binascii import hexlify, unhexlify
import traceback
import sys
from typing import NamedTuple, Any, Optional, Dict, Union, List, Tuple, TYPE_CHECKING

from electrum.util import bfh, bh2u, UserCancelled, UserFacingException
from electrum.bip32 import BIP32Node
from electrum import constants
from electrum.i18n import _
from electrum.transaction import Transaction, PartialTransaction, PartialTxInput, PartialTxOutput
from electrum.keystore import Hardware_KeyStore
from electrum.plugin import Device
from electrum.base_wizard import ScriptTypeNotSupported

from ..hw_wallet import HW_PluginBase
from ..hw_wallet.plugin import (is_any_tx_output_on_change_branch, trezor_validate_op_return_output_and_get_data,
                                get_xpubs_and_der_suffixes_from_txinout)

if TYPE_CHECKING:
    import usb1
    from .client import KeepKeyClient


# TREZOR initialization methods
TIM_NEW, TIM_RECOVER, TIM_MNEMONIC, TIM_PRIVKEY = range(0, 4)


class KeepKey_KeyStore(Hardware_KeyStore):
    hw_type = 'keepkey'
    device = 'KeepKey'

    plugin: 'KeepKeyPlugin'

    def get_client(self, force_pair=True):
        return self.plugin.get_client(self, force_pair)

    def decrypt_message(self, sequence, message, password):
        raise UserFacingException(_('Encryption and decryption are not implemented by {}').format(self.device))

    def sign_message(self, sequence, message, password):
        client = self.get_client()
        address_path = self.get_derivation_prefix() + "/%d/%d"%sequence
        address_n = client.expand_path(address_path)
        msg_sig = client.sign_message(self.plugin.get_coin_name(), address_n, message)
        return msg_sig.signature

    def sign_transaction(self, tx, password):
        if tx.is_complete():
            return
        # previous transactions used as inputs
        prev_tx = {}
        for txin in tx.inputs():
            tx_hash = txin.prevout.txid.hex()
            if txin.utxo is None and not Transaction.is_segwit_input(txin):
                raise UserFacingException(_('Missing previous tx for legacy input.'))
            prev_tx[tx_hash] = txin.utxo

        self.plugin.sign_transaction(self, tx, prev_tx)


class KeepKeyPlugin(HW_PluginBase):
    # Derived classes provide:
    #
    #  class-static variables: client_class, firmware_URL, handler_class,
    #     libraries_available, libraries_URL, minimum_firmware,
    #     wallet_class, ckd_public, types, HidTransport

    firmware_URL = 'https://www.keepkey.com'
    libraries_URL = 'https://github.com/keepkey/python-keepkey'
    minimum_firmware = (1, 0, 0)
    keystore_class = KeepKey_KeyStore
    SUPPORTED_XTYPES = ('standard', 'p2wpkh-p2sh', 'p2wpkh', 'p2wsh-p2sh', 'p2wsh')

    MAX_LABEL_LEN = 32

    def __init__(self, parent, config, name):
        HW_PluginBase.__init__(self, parent, config, name)

        try:
            from . import client
            import keepkeylib
            import keepkeylib.ckd_public
            import keepkeylib.transport_hid
            import keepkeylib.transport_webusb
            self.client_class = client.KeepKeyClient
            self.ckd_public = keepkeylib.ckd_public
            self.types = keepkeylib.client.types
            self.DEVICE_IDS = (keepkeylib.transport_hid.DEVICE_IDS +
                               keepkeylib.transport_webusb.DEVICE_IDS)
            # only "register" hid device id:
            self.device_manager().register_devices(keepkeylib.transport_hid.DEVICE_IDS)
            # for webusb transport, use custom enumerate function:
            self.device_manager().register_enumerate_func(self.enumerate)
            self.libraries_available = True
        except ImportError:
            self.libraries_available = False

    def enumerate(self):
        from keepkeylib.transport_webusb import WebUsbTransport
        results = []
        for dev in WebUsbTransport.enumerate():
            path = self._dev_to_str(dev)
            results.append(Device(path=path,
                                  interface_number=-1,
                                  id_=path,
                                  product_key=(dev.getVendorID(), dev.getProductID()),
                                  usage_page=0,
                                  transport_ui_string=f"webusb:{path}"))
        return results

    @staticmethod
    def _dev_to_str(dev: "usb1.USBDevice") -> str:
        return ":".join(str(x) for x in ["%03i" % (dev.getBusNumber(),)] + dev.getPortNumberList())

    def hid_transport(self, pair):
        from keepkeylib.transport_hid import HidTransport
        return HidTransport(pair)

    def webusb_transport(self, device):
        from keepkeylib.transport_webusb import WebUsbTransport
        for dev in WebUsbTransport.enumerate():
            if device.path == self._dev_to_str(dev):
                return WebUsbTransport(dev)

    def _try_hid(self, device):
        self.logger.info("Trying to connect over USB...")
        if device.interface_number == 1:
            pair = [None, device.path]
        else:
            pair = [device.path, None]

        try:
            return self.hid_transport(pair)
        except BaseException as e:
            # see fdb810ba622dc7dbe1259cbafb5b28e19d2ab114
            # raise
            self.logger.info(f"cannot connect at {device.path} {e}")
            return None

    def _try_webusb(self, device):
        self.logger.info("Trying to connect over WebUSB...")
        try:
            return self.webusb_transport(device)
        except BaseException as e:
            self.logger.info(f"cannot connect at {device.path} {e}")
            return None
{'fraction_non_alphanumeric': 0.0817344589409056, 'fraction_numerical': 0.00844205679201842, 'mean_word_length': 2.702059659090909, 'pattern_counts': {'":': 1, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6014607', 'n_tokens_mistral': 3144, 'n_tokens_neox': 3075, 'n_words': 694}