Added signer authorization generation option that uses standard Ethereum app

middleware/admin/dongle_eth.py

@@ -0,0 +1,143 @@
+# The MIT License (MIT)
+#
+# Copyright (c) 2021 RSK Labs Ltd
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy of
+# this software and associated documentation files (the "Software"), to deal in
+# the Software without restriction, including without limitation the rights to
+# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+# of the Software, and to permit persons to whom the Software is furnished to do
+# so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import ecdsa
+import struct
+
+from enum import IntEnum
+from ledgerblue.comm import getDongle
+from ledgerblue.commException import CommException
+
+
+class DongleEthError(RuntimeError):
+    pass
+
+
+class DongleEthWrongApp(RuntimeError):
+    @staticmethod
+    def is_wrong_app(e):
+        if type(e) == CommException and e.sw == 0x6511:
+            return True
+        return False
+
+
+class DongleEthInvalidPath(RuntimeError):
+    @staticmethod
+    def is_invalid_path(e):
+        if type(e) == CommException and e.sw == 0x6a15:
+            return True
+        return False
+
+
+class DongleEthLocked(RuntimeError):
+    @staticmethod
+    def is_locked(e):
+        if type(e) == CommException and e.sw == 0x6b0c:
+            return True
+        return False
+
+
+# Dongle commands
+class _Command(IntEnum):
+    GET_PUBLIC_ADDRESS = 0x02,
+    SIGN_PERSONAL_MSG = 0x08
+
+
+class _Offset(IntEnum):
+    PUBKEY = 1
+    SIG_R = 1
+    SIG_S = 33
+    SIG_S_END = 65
+
+
+# Handles low-level communication with an ledger device running Ethereum App
+class DongleEth:
+    # APDU prefix
+    CLA = 0xE0
+
+    # Enumeration shorthands
+    CMD = _Command
+    OFF = _Offset
+
+    # Maximum size of msg allowed by sign command
+    MAX_MSG_LEN = 255
+
+    def __init__(self, debug):
+        self.debug = debug
+
+    # Connect to the dongle
+    def connect(self):
+        try:
+            self.dongle = getDongle(self.debug)
+        except CommException as e:
+            msg = "Error connecting: %s" % e.message
+            raise DongleEthError(msg)
+
+    # Disconnect from dongle
+    def disconnect(self):
+        try:
+            if self.dongle and self.dongle.opened:
+                self.dongle.close()
+        except CommException as e:
+            msg = "Error disconnecting: %s" % e.message
+            raise DongleEthError(msg)
+
+    def get_pubkey(self, path):
+        # Skip length byte
+        dongle_path = path.to_binary("big")[1:]
+        result = self._send_command(self.CMD.GET_PUBLIC_ADDRESS,
+                                    bytes([0x00, 0x00, len(dongle_path) + 1,
+                                           len(path.elements)]) + dongle_path)
+        pubkey = result[self.OFF.PUBKEY:self.OFF.PUBKEY + result[0]]
+        return bytes(pubkey)
+
+    def sign(self, path, msg):
+        if len(msg) > self.MAX_MSG_LEN:
+            raise DongleEthError("Message greater than maximum supported size of "
+                                 f"{self.MAX_MSG_LEN} bytes")
+
+        # Skip length byte
+        dongle_path = path.to_binary("big")[1:]
+        encoded_tx = struct.pack(">I", len(msg)) + msg
+        result = self._send_command(self.CMD.SIGN_PERSONAL_MSG,
+                                    bytes([0x00, 0x00,
+                                           len(dongle_path) + 1 + len(encoded_tx),
+                                           len(path.elements)])
+                                    + dongle_path + encoded_tx)
+
+        r = result[self.OFF.SIG_R:self.OFF.SIG_S].hex()
+        s = result[self.OFF.SIG_S:self.OFF.SIG_S_END].hex()
+
+        return ecdsa.util.sigencode_der(int(r, 16), int(s, 16), 0)
+
+    def _send_command(self, cmd, data):
+        try:
+            apdu = bytes([self.CLA, cmd]) + data
+            return self.dongle.exchange(apdu)
+        except (CommException, BaseException) as e:
+            if DongleEthWrongApp.is_wrong_app(e):
+                raise DongleEthWrongApp("Ethereum app not open")
+            if DongleEthInvalidPath.is_invalid_path(e):
+                raise DongleEthInvalidPath("Invalid path for Ethereum app")
+            if DongleEthLocked.is_locked(e):
+                raise DongleEthLocked("Device locked")
+            raise DongleEthError("Error sending command: %s" % str(e))

middleware/admin/misc.py

@@ -26,6 +26,7 @@
 from ledger.hsm2dongle import HSM2Dongle
 from ledger.pin import BasePin
 from .dongle_admin import DongleAdmin
+from .dongle_eth import DongleEth
 
 PIN_ERROR_MESSAGE = ("Invalid pin given. It must be exactly 8 alphanumeric "
                      "characters with at least one alphabetic character.")
@@ -90,6 +91,23 @@ def dispose_hsm(hsm):
     info("OK")
 
 
+def get_eth_dongle(debug):
+    info("Connecting to Ethereum App... ", False)
+    eth = DongleEth(debug)
+    eth.connect()
+    info("OK")
+    return eth
+
+
+def dispose_eth_dongle(eth):
+    if eth is None:
+        return
+
+    info("Disconnecting from Ethereum App... ", False)
+    eth.disconnect()
+    info("OK")
+
+
 def ask_for_pin(require_alpha):
     pin = None
     while pin is None or not BasePin.is_valid(pin, require_alpha=require_alpha):

middleware/admin/signer_authorization.py

@@ -121,9 +121,12 @@ def hash(self):
     def iteration(self):
         return self._iteration
 
+    @property
+    def msg(self):
+        return f"RSK_powHSM_signer_{self._hash}_iteration_{str(self._iteration)}"
+
     def get_authorization_msg(self):
-        msg = f"RSK_powHSM_signer_{self._hash}_iteration_{str(self._iteration)}"
-        return encode_eth_message(msg)
+        return encode_eth_message(self.msg)
 
     def get_authorization_digest(self):
         return sha3.keccak_256(self.get_authorization_msg()).digest()

middleware/comm/bip32.py

@@ -109,10 +109,15 @@ def __init__(self, spec, nelements=5):
     def elements(self):
         return self._elements
 
-    def to_binary(self):
-        binary = struct.pack("<B", len(self._elements))
+    def to_binary(self, byteorder="little"):
+        if byteorder == "big":
+            order_sign = ">"
+        else:
+            order_sign = "<"
+
+        binary = struct.pack(f"{order_sign}B", len(self._elements))
         for element in self.elements:
-            binary += struct.pack("<I", element.index)
+            binary += struct.pack(f"{order_sign}I", element.index)
         return binary
 
     def __str__(self):

middleware/signapp.py

@@ -25,14 +25,22 @@
 from argparse import ArgumentParser
 import logging
 import ecdsa
-from admin.misc import get_hsm, dispose_hsm, info, AdminError
+from admin.misc import (
+    get_hsm,
+    dispose_hsm,
+    get_eth_dongle,
+    dispose_eth_dongle,
+    info,
+    AdminError
+)
 from comm.utils import is_hex_string_of_length
 from comm.bip32 import BIP32Path
 from admin.signer_authorization import SignerAuthorization, SignerVersion
 from admin.ledger_utils import eth_message_to_printable, compute_app_hash
 
 # Default signing path
-DEFAULT_PATH = "m/44'/137'/0'/31/32"
+DEFAULT_LEDGER_PATH = "m/44'/137'/0'/31/32"
+DEFAULT_ETH_PATH = "m/44'/60'/0'/0/0"
 
 # Legacy dongle constants
 COMMAND_SIGN = 0x02
@@ -46,7 +54,7 @@ def main():
 
     parser = ArgumentParser(description="powHSM Signer Authorization Generator")
     parser.add_argument("operation", choices=["hash", "message", "key",
-                                              "ledger", "manual"])
+                                              "ledger", "eth", "manual"])
     parser.add_argument(
         "-s",
         "--signer",
@@ -76,9 +84,8 @@ def main():
         "-p",
         "--path",
         dest="path",
-        help="Path used for signing (only for 'ledger' option). "
-        f"Default \"{DEFAULT_PATH}\"",
-        default=DEFAULT_PATH
+        help="Path used for signing (only for 'ledger' and 'eth' options). "
+        f"Default \"{DEFAULT_LEDGER_PATH}\" (ledger) / \"{DEFAULT_ETH_PATH}\" (eth)"
     )
     parser.add_argument(
         "-g",
@@ -87,6 +94,13 @@ def main():
         help="Signature to add to signer authorization (only for 'manual' option)."
         "Must be a hex-encoded, der-encoded SECP256k1 signature.",
     )
+    parser.add_argument(
+        "-b",
+        "--pubkey",
+        dest="pubkey",
+        action="store_true",
+        help="Retrieve pubkic key (only for 'eth' option)."
+    )
     parser.add_argument(
         "-v",
         "--verbose",
@@ -100,6 +114,14 @@ def main():
 
     try:
         hsm = None
+        eth = None
+
+        # Default path is different for 'ledger' and 'eth' operations
+        if options.path is None:
+            if options.operation == "ledger":
+                options.path = DEFAULT_LEDGER_PATH
+            elif options.operation == "eth":
+                options.path = DEFAULT_ETH_PATH
 
         # Require an output path for certain operations
         if options.operation not in ["hash", "message"] and \
@@ -130,6 +152,26 @@ def main():
 
             # Get dongle access (must be opened in the signer)
             hsm = get_hsm(options.verbose)
+        elif options.operation == "eth":
+            # Parse path
+            path = BIP32Path(options.path)
+
+            # Get dongle access (must have ethereum app open)
+            eth = get_eth_dongle(options.verbose)
+
+            # Retrieve public key
+            info(f"Retrieving public key for path '{str(path)}'...")
+            pubkey = eth.get_pubkey(path)
+            info(f"Public key: {pubkey.hex()}")
+
+            # If options.pubkey is True, we just want to retrieve the public key
+            if options.pubkey:
+                info(f"Opening public key file {options.output_path}...")
+                info("Adding public key...")
+                with open(options.output_path, "w") as file:
+                    file.write(pubkey.hex())
+                info(f"Public key saved to {options.output_path}")
+                sys.exit(0)
 
         # Is there an existing signer authorization? Read it
         signer_authorization = None
@@ -172,7 +214,7 @@ def main():
                                               curve=ecdsa.SECP256k1)
             signature = sk.sign_digest(signer_version.get_authorization_digest(),
                                        sigencode=ecdsa.util.sigencode_der)
-        else:
+        elif options.operation == "ledger":
             # We use private dongle methods to do this, since we don't want
             # to implement legacy signing (i.e., 1.0/1.1) in the HSM2Dongle class
             # Essentially we send 2 messages. The first one with the path and the
@@ -195,6 +237,18 @@ def main():
                     raise Exception()
             except Exception:
                 raise AdminError(f"Bad signature from dongle! (got '{signature.hex}')")
+        elif options.operation == "eth":
+            info("Signing with dongle...")
+            signature = eth.sign(path, signer_version.msg.encode('ascii'))
+            vkey = ecdsa.VerifyingKey.from_string(pubkey, curve=ecdsa.SECP256k1)
+
+            try:
+                if not vkey.verify_digest(
+                        signature, signer_version.get_authorization_digest(),
+                        sigdecode=ecdsa.util.sigdecode_der):
+                    raise Exception()
+            except Exception:
+                raise AdminError(f"Bad signature from dongle! (got '{signature.hex()}')")
 
         # Add the signature to the authorization and save it to disk
         signer_authorization.add_signature(signature.hex())
@@ -206,6 +260,7 @@ def main():
         sys.exit(1)
     finally:
         dispose_hsm(hsm)
+        dispose_eth_dongle(eth)
 
 
 if __name__ == "__main__":