migrate erdpy ledger support from old repo

Author: bogdan-rosianu

README.md

@@ -4,8 +4,6 @@ Elrond Python Command Line Tools and SDK for interacting with the Elrond Network
 ## Documentation
 [docs.elrond.com](https://docs.elrond.com/sdk-and-tools/erdpy/erdpy/)
 
-[pkg.go.dev](https://pkg.go.dev/github.com/ElrondNetwork/elrond-sdk/erdgo)
-
 ## CLI
 [CLI](CLI.md)
 

erdpy/CLI.md

@@ -5,6 +5,7 @@
 from typing import Any, Union
 
 from erdpy import constants, errors, utils
+from erdpy.errors import LedgerError
 from erdpy.interfaces import IAccount, IAddress
 from erdpy.wallet import bech32, generate_pair, pem
 from erdpy.wallet.keyfile import get_password, load_from_key_file
@@ -44,11 +45,13 @@ def get_all(self):
 
 
 class Account(IAccount):
-    def __init__(self, address: Any = None, pem_file: Union[str, None] = None, pem_index: int = 0, key_file: str = "", pass_file: str = ""):
+    def __init__(self, address: Any = None, pem_file: Union[str, None] = None, pem_index: int = 0, key_file: str = "", pass_file: str = "",
+                 ledger: bool = False):
         self.address = Address(address)
         self.pem_file = pem_file
         self.pem_index = int(pem_index)
         self.nonce: int = 0
+        self.ledger = ledger
 
         if pem_file:
             seed, pubkey = pem.parse(self.pem_file, self.pem_index)
@@ -66,6 +69,8 @@ def sync_nonce(self, proxy: Any):
         logger.info(f"Account.sync_nonce() done: {self.nonce}")
 
     def get_seed(self) -> bytes:
+        if self.ledger:
+            raise LedgerError("cannot get seed from a Ledger account")
         return unhexlify(self.private_key_seed)
 
 

erdpy/accounts.py

@@ -13,6 +13,7 @@
 import erdpy.cli_data
 import erdpy.cli_deps
 import erdpy.cli_dispatcher
+import erdpy.cli_ledger
 import erdpy.cli_network
 import erdpy.cli_testnet
 import erdpy.cli_transactions
@@ -85,6 +86,7 @@ def setup_parser():
     commands.append(erdpy.cli_transactions.setup_parser(subparsers))
     commands.append(erdpy.cli_validators.setup_parser(subparsers))
     commands.append(erdpy.cli_accounts.setup_parser(subparsers))
+    commands.append(erdpy.cli_ledger.setup_parser(subparsers))
     commands.append(erdpy.cli_wallet.setup_parser(subparsers))
     commands.append(erdpy.cli_network.setup_parser(subparsers))
     commands.append(erdpy.cli_cost.setup_parser(subparsers))

erdpy/cli.py

@@ -1,11 +1,13 @@
 import argparse
 import ast
+import string
 import sys
 from argparse import FileType
 from typing import Any, List, Text
 
 from erdpy import config, errors, scope, utils
 from erdpy.accounts import Account
+from erdpy.ledger.ledger_functions import do_get_ledger_address
 from erdpy.proxy.core import ElrondProxy
 from erdpy.transactions import Transaction
 
@@ -69,18 +71,22 @@ def add_tx_args(sub: Any, with_nonce: bool = True, with_receiver: bool = True, w
 
     sub.add_argument("--chain", default=scope.get_chain_id(), help="the chain identifier (default: %(default)s)")
     sub.add_argument("--version", type=int, default=scope.get_tx_version(), help="the transaction version (default: %(default)s)")
+    sub.add_argument("--options", type=int, default=0,  help="the transaction options (default: 0)")
 
 
 def add_wallet_args(sub: Any):
-    sub.add_argument("--pem", required=not (utils.is_arg_present("--keyfile", sys.argv)), help="🔑 the PEM file, if keyfile not provided")
+    sub.add_argument("--pem", required=check_if_sign_method_required("--pem"), help="🔑 the PEM file, if keyfile not provided")
     sub.add_argument("--pem-index", default=0, help="🔑 the index in the PEM file (default: %(default)s)")
-    sub.add_argument("--keyfile", required=not (utils.is_arg_present("--pem", sys.argv)), help="🔑 a JSON keyfile, if PEM not provided")
-    sub.add_argument("--passfile", required=not (utils.is_arg_present("--pem", sys.argv)), help="🔑 a file containing keyfile's password, if keyfile provided")
+    sub.add_argument("--keyfile", required=check_if_sign_method_required("--keyfile"), help="🔑 a JSON keyfile, if PEM not provided")
+    sub.add_argument("--passfile", required=(utils.is_arg_present("--keyfile", sys.argv)), help="🔑 a file containing keyfile's password, if keyfile provided")
+    sub.add_argument("--ledger", action="store_true", required=check_if_sign_method_required("--ledger"), default=False, help="🔐 bool flag for signing transaction using ledger")
+    sub.add_argument("--ledger-account-index", type=int, default=0, help="🔐 the index of the account when using Ledger")
+    sub.add_argument("--ledger-address-index", type=int, default=0, help="🔐 the index of the address when using Ledger")
     sub.add_argument("--sender-username", required=False, help="🖄 the username of the sender")
 
 
 def add_proxy_arg(sub: Any):
-    sub.add_argument("--proxy", default=scope.get_proxy(), help="🖧 the URL of the proxy (default: %(default)s)")
+    sub.add_argument("--proxy", default=scope.get_proxy(), help="🔗 the URL of the proxy (default: %(default)s)")
 
 
 def add_outfile_arg(sub: Any, what: str = ""):
@@ -109,6 +115,9 @@ def prepare_nonce_in_args(args: Any):
             account = Account(pem_file=args.pem, pem_index=args.pem_index)
         elif args.keyfile and args.passfile:
             account = Account(key_file=args.keyfile, pass_file=args.passfile)
+        elif args.ledger:
+            address = do_get_ledger_address(account_index=args.ledger_account_index, address_index=args.ledger_address_index)
+            account = Account(address=address)
         else:
             raise errors.NoWalletProvided()
 
@@ -138,3 +147,17 @@ def send_or_simulate(tx: Transaction, args: Any):
     elif args.simulate:
         response = tx.simulate(ElrondProxy(args.proxy))
         utils.dump_out_json(response)
+
+
+def check_if_sign_method_required(checked_method: string) -> bool:
+    methods = ["--pem", "--keyfile", "--ledger"]
+    rest_of_methods = []
+    for method in methods:
+        if method != checked_method:
+            rest_of_methods.append(method)
+
+    for method in rest_of_methods:
+        if utils.is_arg_present(method, sys.argv):
+            return False
+
+    return True

erdpy/cli_shared.py

@@ -2,6 +2,7 @@
 from typing import Any
 
 from erdpy import cli_shared, utils
+from erdpy.ledger.ledger_functions import do_get_ledger_address
 from erdpy.proxy.core import ElrondProxy
 from erdpy.transactions import Transaction, do_prepare_transaction
 
@@ -51,6 +52,9 @@ def create_transaction(args: Any):
     cli_shared.check_broadcast_args(args)
     cli_shared.prepare_nonce_in_args(args)
 
+    if args.ledger:
+        args.ledger_address = do_get_ledger_address(account_index=args.ledger_account_index, address_index=args.ledger_address_index)
+
     if args.data_file:
         args.data = utils.read_file(args.data_file)
 

erdpy/cli_transactions.py

@@ -178,3 +178,8 @@ def __init__(self):
 class TestnetError(KnownError):
     def __init__(self, message: str):
         super().__init__(message)
+
+
+class LedgerError(KnownError):
+    def __init__(self, message: str):
+        super().__init__("Ledger error: " + message)

erdpy/errors.py

@@ -0,0 +1,41 @@
+import string
+
+
+def load_ledger_config_from_response(response: bytes):
+    config = ElrondLedgerAppConfiguration()
+
+    config.data_activated = False
+    if response[0] == 0x01:
+        config.data_activated = True
+
+    config.account_index = response[1]
+    config.address_index = response[2]
+
+    version = str(response[3]) + "." + str(response[4]) + "." + str(response[5])
+    config.version = version
+
+    return config
+
+
+def compare_versions(version1: string, version2: string) -> int:
+    version1_tuple = version_tuple(version1)
+    version2_tuple = version_tuple(version2)
+    if version1_tuple == version2_tuple:
+        return 0
+    if version1_tuple < version2_tuple:
+        return -1
+    return 1
+
+
+def version_tuple(v):
+    filled = []
+    for point in v.split("."):
+        filled.append(point.zfill(8))
+    return tuple(filled)
+
+
+class ElrondLedgerAppConfiguration:
+    data_activated: bool
+    account_index: int
+    address_index: int
+    version: string

erdpy/ledger/__init__.py

@@ -0,0 +1,140 @@
+from ledgercomm import Transport
+
+from erdpy.errors import LedgerError
+from erdpy.ledger.config import load_ledger_config_from_response, ElrondLedgerAppConfiguration
+
+SIGN_USING_HASH_VERSION = "1.0.11"
+
+
+class Apdu:
+    cla: int
+    ins: int
+    p1: int
+    p2: int
+    data: bytes
+
+
+class ElrondLedgerApp:
+    def __init__(self):
+        self.transport = Transport(interface="hid", debug=False)  # Nano S/X using HID interface
+
+    def close(self):
+        self.transport.close()
+
+    def set_address(self, account_index=0, address_index=0):
+        data = account_index.to_bytes(4, byteorder='big') + address_index.to_bytes(4, byteorder='big')
+        self.transport.send(cla=0xed, ins=0x05, p1=0x00, p2=0x00, cdata=data)
+        sw, response = self.transport.recv()
+        err = get_error(sw)
+        if err != '':
+            raise LedgerError(err)
+
+    def get_address(self, account_index=0, address_index=0):
+        data = account_index.to_bytes(4, byteorder='big') + address_index.to_bytes(4, byteorder='big')
+        self.transport.send(cla=0xed, ins=0x03, p1=0x00, p2=0x00, cdata=data)
+        sw, response = self.transport.recv()
+        err = get_error(sw)
+        if err != '':
+            raise LedgerError(err)
+
+        address = response[1:].decode("utf-8")
+        return address
+
+    def get_app_configuration(self) -> ElrondLedgerAppConfiguration:
+        self.transport.send(cla=0xed, ins=0x02, p1=0x00, p2=0x00, cdata=b"")
+        sw, response = self.transport.recv()
+        err = get_error(sw)
+        if err != '':
+            raise LedgerError(err)
+        return load_ledger_config_from_response(response)
+
+    def get_version(self):
+        config = self.get_app_configuration()
+        return config.version
+
+    def sign_transaction(self, marshaled_tx: bytes, should_use_hash_signing: bool):
+        total_size = len(marshaled_tx)
+        max_chunk_size = 150
+
+        apdus = list()
+
+        offset = 0
+        while offset != total_size:
+            is_first = offset == 0
+
+            apdu = Apdu()
+
+            if is_first:
+                apdu.p1 = 0x00
+            else:
+                apdu.p1 = 0x80
+
+            has_more = offset + max_chunk_size < total_size
+            chunk_size = total_size - offset
+            if has_more:
+                chunk_size = max_chunk_size
+
+            ins_signing_method = 0x04
+            if should_use_hash_signing:
+                ins_signing_method = 0x07
+
+            apdu.ins = ins_signing_method
+            apdu.p2 = 0x00
+            apdu.cla = 0xed
+            apdu.data = marshaled_tx[offset:offset + chunk_size]
+
+            apdus.append(apdu)
+
+            offset += chunk_size
+
+        return self.get_signature_from_apdus(apdus)
+
+    def get_signature_from_apdus(self, apdus):
+        sw = []
+        response = []
+        for apdu in apdus:
+            self.transport.send(
+                cla=apdu.cla,
+                ins=apdu.ins,
+                p1=apdu.p1,
+                p2=apdu.p2,
+                cdata=apdu.data)
+            sw, response = self.transport.recv()
+
+        if len(response) != 65 or response[0] != 64 or get_error(sw) != '':
+            err_message = "signature failed"
+            err = get_error(sw)
+            if err != '':
+                err_message += ': ' + err
+            raise LedgerError(err_message)
+
+        signature = response[1:len(response)].hex()
+        return signature
+
+
+def get_error(code):
+    switcher = {
+        0x9000: '',
+        0x6985: 'user denied',
+        0x6D00: 'unknown instruction',
+        0x6E00: 'wrong cla',
+        0x6E10: 'signature failed',
+        0x6E01: 'invalid arguments',
+        0x6E02: 'invalid message',
+        0x6E03: 'invalid p1',
+        0x6E04: 'message too long',
+        0x6E05: 'receiver too long',
+        0x6E06: 'amount too long',
+        0x6E07: 'contract data disabled',
+        0x6E08: 'message incomplete',
+        0x6E09: 'wrong tx version',
+        0x6E0A: 'nonce too long',
+        0x6E0B: 'invalid amount',
+        0x6E0C: 'invalid fee',
+        0x6E0D: 'pretty failed',
+        0x6E0E: 'data too long',
+        0x6E0F: 'wrong tx options',
+        0x6E11: 'regular signing is deprecated',
+    }
+
+    return switcher.get(code, "unknown error code: " + hex(code))