Implemented SGX manager

- Added password changing capabilities to the SGX powHSM implementation
- Added main SGX manager script
- Added build script for the SGX manager
- Added password change and password retries getter commands to the HSM2DongleSGX
- Updated user options to make names more generic
- Added unit tests
- Updated middleware docker do script to expose the host to the container (allowing for running tests against e.g. an SGX powHSM simulator)

docker/mware/do

@@ -29,4 +29,4 @@ WORKDIR=$1; shift
 BINARY=$1; shift
 ARGS=$@
 
-docker run -ti --rm --name hsm-mware -v $HSM_ROOT:/hsm2 -v /dev/bus/usb:/dev/bus/usb --privileged -p$PORT:$PORT -w $WORKDIR $DOCKER_IMAGE $BINARY $ARGS
+docker run -ti --rm --name hsm-mware -v $HSM_ROOT:/hsm2 -v /dev/bus/usb:/dev/bus/usb --privileged --add-host=host.docker.internal:host-gateway -p$PORT:$PORT -w $WORKDIR $DOCKER_IMAGE $BINARY $ARGS

firmware/src/powhsm/src/err.h

@@ -35,6 +35,7 @@ typedef enum {
     ERR_DEVICE_ONBOARDED = 0x6BEF,
     ERR_ONBOARDING = 0x6BF0,
     ERR_DEVICE_LOCKED = 0x6BF1,
+    ERR_PASSWORD_CHANGE = 0x6BF2,
 } err_code_signer_t;
 
 #endif // __ERR_H

firmware/src/powhsm/src/instructions.h

@@ -58,6 +58,7 @@ typedef enum {
     SGX_RETRIES = 0xA2,
     SGX_UNLOCK = 0xA3,
     SGX_ECHO = 0xA4,
+    SGX_CHANGE_PASSWORD = 0xA5,
 } apdu_instruction_t;
 
 #endif // __INSTRUCTIONS_H

firmware/src/sgx/src/trusted/system.c

@@ -67,6 +67,24 @@ static unsigned int do_onboard(unsigned int rx) {
     return TX_NO_DATA();
 }
 
+static unsigned int do_change_password(unsigned int rx) {
+    // Require a nonblank password
+    if (APDU_DATA_SIZE(rx) < 1) {
+        THROW(ERR_INVALID_DATA_SIZE);
+    }
+
+    // Password change
+    uint8_t tmp_buffer[apdu_buffer_size];
+    size_t password_length = APDU_DATA_SIZE(rx);
+    memcpy(tmp_buffer, APDU_DATA_PTR, password_length);
+    if (!access_set_password((char*)tmp_buffer, password_length)) {
+        THROW(ERR_PASSWORD_CHANGE);
+    }
+
+    SET_APDU_OP(1);
+    return TX_NO_DATA();
+}
+
 static unsigned int do_unlock(unsigned int rx) {
     if (!access_is_locked()) {
         SET_APDU_OP(1);
@@ -124,6 +142,11 @@ static external_processor_result_t system_do_process_apdu(unsigned int rx) {
     case SGX_ECHO:
         result.tx = do_echo(rx);
         break;
+    case SGX_CHANGE_PASSWORD:
+        REQUIRE_ONBOARDED();
+        REQUIRE_UNLOCKED();
+        result.tx = do_change_password(rx);
+        break;
     default:
         result.handled = false;
     }

middleware/build/manager_sgx

@@ -0,0 +1,2 @@
+#!/bin/bash
+source $(dirname $0)/bld-docker manager_sgx

middleware/manager_sgx.py

@@ -0,0 +1,61 @@
+# 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 os
+from sgx.hsm2dongle import HSM2DongleSGX
+from mgr.runner import ManagerRunner
+from ledger.pin import FileBasedPin
+from user.options import UserOptionParser
+
+
+def load_pin(user_options):
+    env_pin = os.environ.get("PIN", None)
+    if env_pin is not None:
+        env_pin = env_pin.encode()
+    pin = FileBasedPin(
+        user_options.pin_file,
+        default_pin=env_pin,
+        force_change=user_options.force_pin_change,
+    )
+    return pin
+
+
+def configure_protocol_messages(protocol):
+    protocol.MESSAGES = {
+        "restart": "restart the SGX powHSM",
+    }
+
+
+if __name__ == "__main__":
+    user_options = UserOptionParser("Start the powHSM manager for SGX",
+                                    with_pin=True,
+                                    with_tcpconn=True,
+                                    host_name="SGX",
+                                    default_tcpconn_port=7777).parse()
+
+    runner = ManagerRunner("powHSM manager for SGX",
+                           lambda options: HSM2DongleSGX(options.tcpconn_host,
+                                                         options.tcpconn_port,
+                                                         options.io_debug),
+                           load_pin, configure_protocol_messages)
+
+    runner.run(user_options)

middleware/sgx/hsm2dongle.py

@@ -25,8 +25,10 @@
 
 
 class SgxCommand(IntEnum):
+    SGX_RETRIES = 0xA2,
     SGX_UNLOCK = 0xA3,
     SGX_ECHO = 0xA4,
+    SGX_CHANGE_PASSWORD = 0xA5,
 
 
 class HSM2DongleSGX(HSM2DongleTCP):
@@ -44,3 +46,15 @@ def unlock(self, pin):
 
         # Nonzero indicates device unlocked
         return response[2] != 0
+
+    # change pin
+    def new_pin(self, pin):
+        response = self._send_command(SgxCommand.SGX_CHANGE_PASSWORD, bytes([0]) + pin)
+
+        # One indicates pin changed
+        return response[2] == 1
+
+    # returns the number of pin retries available
+    def get_retries(self):
+        apdu_rcv = self._send_command(SgxCommand.SGX_RETRIES)
+        return apdu_rcv[2]

middleware/tests/sgx/test_hsm2dongle.py

@@ -0,0 +1,92 @@
+# 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.
+
+from unittest import TestCase
+from unittest.mock import Mock, patch
+from sgx.hsm2dongle import HSM2DongleSGX
+from ledger.hsm2dongle import HSM2DongleError
+
+import logging
+
+logging.disable(logging.CRITICAL)
+
+
+class TestHSM2DongleSGX(TestCase):
+    EXPECTED_DONGLE_TIMEOUT = 10
+
+    @patch("ledger.hsm2dongle_tcp.getDongle")
+    def setUp(self, getDongleMock):
+        self.dongle = Mock()
+        self.getDongleMock = getDongleMock
+        self.getDongleMock.return_value = self.dongle
+        self.hsm2dongle = HSM2DongleSGX("a-host", 1234, "a-debug-value")
+
+        self.getDongleMock.assert_not_called()
+        self.hsm2dongle.connect()
+        self.getDongleMock.assert_called_with("a-host", 1234, "a-debug-value")
+        self.assertEqual(self.hsm2dongle.dongle, self.dongle)
+
+    def assert_exchange_called(self, bs):
+        self.dongle.exchange.assert_called_with(bs, timeout=self.EXPECTED_DONGLE_TIMEOUT)
+
+    def test_echo_ok(self):
+        self.dongle.exchange.return_value = bytes([0x80, 0xA4, 0x41, 0x42, 0x43])
+        self.assertTrue(self.hsm2dongle.echo())
+        self.assert_exchange_called(bytes([0x80, 0xA4, 0x41, 0x42, 0x43]))
+
+    def test_echo_response_differs(self):
+        self.dongle.exchange.return_value = bytes([1, 2, 3])
+        self.assertFalse(self.hsm2dongle.echo())
+        self.assert_exchange_called(bytes([0x80, 0xA4, 0x41, 0x42, 0x43]))
+
+    def test_echo_error_triggered(self):
+        self.dongle.exchange.side_effect = RuntimeError("SomethingWentWrong")
+        with self.assertRaises(HSM2DongleError) as cm:
+            self.hsm2dongle.echo()
+        self.assert_exchange_called(bytes([0x80, 0xA4, 0x41, 0x42, 0x43]))
+
+    def test_unlock_ok(self):
+        self.dongle.exchange.return_value = bytes([0xAA, 0xBB, 0xCC, 0xDD])
+        self.assertTrue(self.hsm2dongle.unlock(b'a-password'))
+        self.assert_exchange_called(bytes([0x80, 0xA3, 0x00]) + b'a-password')
+
+    def test_unlock_wrong_pass(self):
+        self.dongle.exchange.return_value = bytes([0xAA, 0xBB, 0x00, 0xDD])
+        self.assertFalse(self.hsm2dongle.unlock(b'wrong-pass'))
+        self.assert_exchange_called(bytes([0x80, 0xA3, 0x00]) + b'wrong-pass')
+
+    def test_newpin_ok(self):
+        self.dongle.exchange.return_value = bytes([0xAA, 0xBB, 0x01, 0xDD])
+        self.assertTrue(self.hsm2dongle.new_pin(b'new-password'))
+        self.assert_exchange_called(bytes([0x80, 0xA5, 0x00]) + b'new-password')
+
+    def test_newpin_error(self):
+        self.dongle.exchange.return_value = bytes([0xAA, 0xBB, 0x55, 0xDD])
+        self.assertFalse(self.hsm2dongle.new_pin(b'new-password'))
+        self.assert_exchange_called(bytes([0x80, 0xA5, 0x00]) + b'new-password')
+
+    def test_get_retries(self):
+        self.dongle.exchange.return_value = bytes([0xAA, 0xBB, 0x05, 0xDD])
+        self.assertEqual(5, self.hsm2dongle.get_retries())
+        self.assert_exchange_called(bytes([0x80, 0xA2]))
+
+

middleware/user/options.py

@@ -28,23 +28,25 @@ def __init__(
         self,
         description,
         with_pin,
-        with_tcpsigner=False,
+        with_tcpconn=False,
+        host_name="",
         default_port=9999,
         default_host="localhost",
         default_pin_file="pin.txt",
         default_logging_config_path="logging.cfg",
-        default_tcpsigner_host="localhost",
-        default_tcpsigner_port=8888,
+        default_tcpconn_host="localhost",
+        default_tcpconn_port=8888,
     ):
         self.description = description
         self.with_pin = with_pin
-        self.with_tcpsigner = with_tcpsigner
+        self.with_tcpconn = with_tcpconn
+        self.host_name = host_name
         self.default_port = default_port
         self.default_host = default_host
         self.default_pin_file = default_pin_file
         self.default_logging_config_path = default_logging_config_path
-        self.default_tcpsigner_port = default_tcpsigner_port
-        self.default_tcpsigner_host = default_tcpsigner_host
+        self.default_tcpconn_port = default_tcpconn_port
+        self.default_tcpconn_host = default_tcpconn_host
 
     def parse(self):
         parser = ArgumentParser(description=self.description)
@@ -65,10 +67,10 @@ def parse(self):
         )
         parser.add_argument(
             "-D",
-            "--dongledebug",
-            dest="dongle_debug",
+            "--iodebug",
+            dest="io_debug",
             action="store_true",
-            help="Low level dongle debug. (defaults to no)",
+            help="Low level I/O debug. (defaults to no)",
         )
 
         if self.with_pin:
@@ -102,21 +104,22 @@ def parse(self):
             help="Run in version 1 mode. (defaults to no)",
         )
 
-        if self.with_tcpsigner:
+        if self.with_tcpconn:
             parser.add_argument(
-                "-tp",
-                "--tcpsigner-port",
-                dest="tcpsigner_port",
-                help=f"TCPSigner listening port (default {self.default_tcpsigner_port})",
+                f"-{self.host_name.lower()[0]}p",
+                f"--{self.host_name.lower()}-port",
+                dest="tcpconn_port",
+                help=f"{self.host_name} listening port (default "
+                     f"{self.default_tcpconn_port})",
                 type=int,
-                default=self.default_tcpsigner_port,
+                default=self.default_tcpconn_port,
             )
             parser.add_argument(
-                "-th",
-                "--tcpsigner-host",
-                dest="tcpsigner_host",
-                help=f"TCPSigner host. (default '{self.default_tcpsigner_host}')",
-                default=self.default_tcpsigner_host,
+                f"-{self.host_name.lower()[0]}h",
+                f"--{self.host_name.lower()}-host",
+                dest="tcpconn_host",
+                help=f"{self.host_name} host. (default '{self.default_tcpconn_host}')",
+                default=self.default_tcpconn_host,
             )
 
         options = parser.parse_args()