Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .flake8
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
[flake8]
exclude = *.pyc,__pycache__,hwilib/devices/ledger_bitcoin/,hwilib/devices/btchip,hwilib/devices/ckcc/,hwilib/devices/jadepy/,hwilib/devices/trezorlib/,test/work/,hwilib/ui,hwilib/devices/bitbox02_lib
exclude = *.pyc,__pycache__,hwilib/devices/ledger_bitcoin/,hwilib/devices/btchip,hwilib/devices/ckcc/,hwilib/devices/jadepy/,hwilib/devices/trezorlib/,test/work/,hwilib/ui,hwilib/devices/bitbox02_lib,hwilib/devices/cypherock_sdk/app_btc/proto/generated/,hwilib/devices/cypherock_sdk/app_manager/proto/generated/,hwilib/devices/cypherock_sdk/core/encoders/proto/generated/
ignore = E261,E302,E305,E501,E722,W5,E231
per-file-ignores = setup.py:E122
40 changes: 14 additions & 26 deletions hwilib/devices/cypherock.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,8 @@
Chain,
)

import hid

CYPHEROCK_VENDOR_ID = 0x3503
CYPHEROCK_PRODUCT_ID = 0x0103
from .cypherock_sdk.hw_hid import get_available_devices, DeviceConnection
from .cypherock_sdk.interfaces import IDevice

class CypherockClient(HardwareWalletClient):
"""
Expand All @@ -35,14 +33,21 @@ class CypherockClient(HardwareWalletClient):
def __init__(self, path: str, password: Optional[str] = None, expert: bool = False, chain: Chain = Chain.MAIN) -> None:
super(CypherockClient, self).__init__(path, password, expert, chain)

all_devices = get_available_devices()
for device in all_devices:
if device["path"] == path:
self.device: IDevice = device
break
else:
raise DeviceConnectionError(f"Device not found: {path}")

try:
self.device = hid.device()
self.device.open_path(path.encode())
self.connection = DeviceConnection.connect(self.device)
except Exception as e:
raise DeviceConnectionError(f"Failed to connect to Cypherock X1: {e}")

def get_master_fingerprint(self) -> bytes:
return bytes.fromhex(self.device.get_serial_number_string())
return bytes.fromhex(self.device["fingerprint"])

def get_pubkey_at_path(self, path: str) -> ExtendedKey:
raise NotImplementedError("The CypherockClient class "
Expand Down Expand Up @@ -105,7 +110,7 @@ def backup_device(self, label: str = "", passphrase: str = "") -> bool:
raise UnavailableActionError('The Cypherock X1 does not support creating a backup via software')

def close(self) -> None:
self.device.close()
self.connection.destroy()

def prompt_pin(self) -> bool:
"""
Expand Down Expand Up @@ -142,21 +147,4 @@ def can_sign_taproot(self) -> bool:


def enumerate(password: Optional[str] = None, expert: bool = False, chain: Chain = Chain.MAIN, allow_emulators: bool = False) -> List[Dict[str, Any]]:
results = []
devices = []
devices.extend(hid.enumerate(CYPHEROCK_VENDOR_ID, CYPHEROCK_PRODUCT_ID))

for d in devices:
if d.get('path') is not None and d.get('serial_number') is not None:
d_data: Dict[str, Any] = {}
d_data['type'] = 'cypherock'
d_data['model'] = 'cypherock-x1'
d_data['label'] = None
d_data['needs_pin_sent'] = False
d_data['needs_passphrase_sent'] = False
d_data['path'] = d['path'].decode()
d_data['fingerprint'] = bytes.fromhex(d['serial_number']).hex() # Using the serial number as the fingerprint for now

results.append(d_data)

return results
return get_available_devices()
3 changes: 3 additions & 0 deletions hwilib/devices/cypherock_sdk/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
# Cypherock Python SDK

This is a modified/stripped down version of the [Cypherock Python SDK](https://github.com/Cypherock/sdk-python) which in itself is the converted version of the official [Cypherock Typescript SDK](https://github.com/Cypherock/sdk) to communicate with the Cypherock X1 firmware.
Empty file.
5 changes: 5 additions & 0 deletions hwilib/devices/cypherock_sdk/app_btc/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
from .app import BtcApp

__all__ = [
"BtcApp",
]
94 changes: 94 additions & 0 deletions hwilib/devices/cypherock_sdk/app_btc/app.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
from ..interfaces import IDeviceConnection
from ..core import SDK
from . import operations


class BtcApp:
"""
Bitcoin application class for Cypherock SDK.
"""

APPLET_ID = 2

def __init__(self, sdk: SDK):
"""
Private constructor. Use create() class method instead.

Args:
sdk: SDK instance
"""
self._sdk = sdk

@classmethod
def create(cls, connection: IDeviceConnection) -> "BtcApp":
"""
Create a new BtcApp instance.

Args:
connection: Device connection instance

Returns:
BtcApp instance
"""
sdk = SDK.create(connection, cls.APPLET_ID)
return cls(sdk)

def get_public_key(
self, params: operations.GetPublicKeyParams
) -> operations.GetPublicKeyResult:
"""
Get public key from device.

Args:
params: Parameters for getting public key

Returns:
Public key result
"""
return self._sdk.run_operation(
lambda: operations.get_public_key(self._sdk, params)
)

def get_xpubs(
self, params: operations.GetXpubsParams
) -> operations.GetXpubsResultResponse:
"""
Get extended public keys from device.

Args:
params: Parameters for getting xpubs

Returns:
Extended public keys result
"""
return self._sdk.run_operation(
lambda: operations.get_xpubs(self._sdk, params)
)

def sign_txn(
self, params: operations.SignTxnParams
) -> operations.SignTxnResult:
"""
Sign Bitcoin transaction on device.

Args:
params: Parameters for signing transaction

Returns:
Sign transaction result
"""
return self._sdk.run_operation(
lambda: operations.sign_txn(self._sdk, params)
)

def destroy(self) -> None:
"""
Destroy the SDK instance and cleanup resources.
"""
return self._sdk.destroy()

def abort(self) -> None:
"""
Send abort signal to device.
"""
self._sdk.send_abort()
33 changes: 33 additions & 0 deletions hwilib/devices/cypherock_sdk/app_btc/operations/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
from .get_public_key import (
get_public_key,
GetPublicKeyEvent,
GetPublicKeyParams,
GetPublicKeyResult,
)
from .get_xpubs import (
get_xpubs,
GetXpubsEvent,
GetXpubsParams,
)
from .types import GetXpubsResultResponse
from .sign_txn import (
sign_txn,
SignTxnEvent,
SignTxnParams,
SignTxnResult,
)

__all__ = [
"get_public_key",
"get_xpubs",
"sign_txn",
"GetPublicKeyEvent",
"GetPublicKeyParams",
"GetPublicKeyResult",
"GetXpubsEvent",
"GetXpubsParams",
"GetXpubsResultResponse",
"SignTxnEvent",
"SignTxnParams",
"SignTxnResult",
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
from ....core.types import ISDK
from ....common_utils import assert_condition, create_status_listener
from ....errors import DeviceAppError, DeviceAppErrorType
from ...proto.generated.btc.get_public_key_pb2 import GetPublicKeyStatus
from ...proto.generated.common_pb2 import SeedGenerationStatus
from ...utils import (
assert_or_throw_invalid_result,
OperationHelper,
assert_derivation_path,
)
from .types import (
GetPublicKeyEvent,
GetPublicKeyParams,
GetPublicKeyResult,
)

__all__ = [
"get_public_key",
"GetPublicKeyEvent",
"GetPublicKeyParams",
"GetPublicKeyResult",
]

import logging
logger = logging.getLogger(__name__)


def get_public_key(
sdk: ISDK,
params: GetPublicKeyParams,
) -> GetPublicKeyResult:
"""
Get public key from device.

Args:
sdk: SDK instance
params: Parameters including wallet_id, derivation_path, and optional on_event handler

Returns:
Result containing public_key

Raises:
AssertionError: If parameters are invalid
"""
assert_condition(params, "params should be defined")
assert_condition(params.wallet_id, "wallet_id should be defined")
assert_derivation_path(params.derivation_path)
assert_condition(
len(params.derivation_path) == 5,
"derivation_path should be of depth 5",
)

status_listener = create_status_listener(
{
"enums": GetPublicKeyEvent,
"operationEnums": GetPublicKeyStatus,
"seedGenerationEnums": SeedGenerationStatus,
"onEvent": params.on_event,
"logger": logger,
}
)
on_status = status_listener["onStatus"]
force_status_update = status_listener["forceStatusUpdate"]

helper = OperationHelper(
sdk=sdk,
query_key="get_public_key",
result_key="get_public_key",
on_status=on_status,
)

helper.send_query(
{
"initiate": {
"wallet_id": params.wallet_id,
"derivation_path": params.derivation_path,
}
}
)

result = helper.wait_for_result()
assert_or_throw_invalid_result(result.result)

force_status_update(GetPublicKeyEvent.VERIFY)

if not result.result.public_key or len(result.result.public_key) == 0:
raise DeviceAppError(DeviceAppErrorType.INVALID_MSG_FROM_DEVICE)

return GetPublicKeyResult(
public_key=result.result.public_key,
address=result.result.address,
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
from typing import Callable, Optional, List
from dataclasses import dataclass
from enum import IntEnum


class GetPublicKeyEvent(IntEnum):
"""Events that can occur during get public key operation."""

INIT = 0
CONFIRM = 1
PASSPHRASE = 2
PIN_CARD = 3
VERIFY = 4


GetPublicKeyEventHandler = Callable[[GetPublicKeyEvent], None]


@dataclass
class GetPublicKeyParams:
"""Parameters for get public key operation."""

wallet_id: bytes
derivation_path: List[int]
on_event: Optional[GetPublicKeyEventHandler] = None


@dataclass
class GetPublicKeyResult:
"""Result of get public key operation."""

public_key: bytes
address: str
Loading