Trezor device authentication

README.md

@@ -18,6 +18,19 @@ pip install afp-sdk
 See [afp.autonity.org](https://afp.autonity.org/) for the Autonomous Futures Protocol
 documentation, including the Python SDK reference.
 
+## Authentication
+
+The SDK supports 3 methods for authenticating with AutEx and signing blockchain
+transactions.
+
+- **Private key:** Use an Autonity account's private key as a hex-string with `0x`
+  prefix.
+- **Key file:** Use a [Geth / Clef](https://geth.ethereum.org/docs/fundamentals/account-management)
+  key file.
+- **Trezor device:** Use a [Trezor](https://trezor.io/) hardware wallet. The derivation
+  path of an Ethereum account needs to be specified, which can be found in the Account
+  Settings in the Trezor Suite.
+
 ## Overview
 
 The `afp` package consists of the following:

afp/__init__.py

@@ -2,7 +2,12 @@
 
 from . import bindings, enums, exceptions, schemas
 from .afp import AFP
-from .auth import Authenticator, KeyfileAuthenticator, PrivateKeyAuthenticator
+from .auth import (
+    Authenticator,
+    KeyfileAuthenticator,
+    PrivateKeyAuthenticator,
+    TrezorAuthenticator,
+)
 from .exceptions import AFPException
 
 __all__ = (
@@ -15,4 +20,5 @@
     "Authenticator",
     "KeyfileAuthenticator",
     "PrivateKeyAuthenticator",
+    "TrezorAuthenticator",
 )

afp/afp.py

@@ -1,4 +1,9 @@
-from .auth import Authenticator, KeyfileAuthenticator, PrivateKeyAuthenticator
+from .auth import (
+    Authenticator,
+    KeyfileAuthenticator,
+    PrivateKeyAuthenticator,
+    TrezorAuthenticator,
+)
 from .config import Config
 from .api.admin import Admin
 from .api.margin_account import MarginAccount
@@ -25,9 +30,10 @@ class AFP:
     ----------
     authenticator : afp.Authenticator, optional
         The default authenticator for signing transactions & messages. Can also be set
-        with environment variables; use `AFP_PRIVATE_KEY` for private key
-        authentication, `AFP_KEYFILE` and `AFP_KEYFILE_PASSWORD` for keyfile
-        authentication.
+        with environment variables: use `AFP_PRIVATE_KEY` for private key
+        authentication; `AFP_KEYFILE` and `AFP_KEYFILE_PASSWORD` for keyfile
+        authentication; `AFP_TREZOR_PATH_OR_INDEX` and `AFP_TREZOR_PASSPHRASE` for
+        Trezor device authentication.
     rpc_url : str, optional
         The URL of an Autonity RPC provider. Can also be set with the `AFP_RPC_URL`
         environment variable.
@@ -217,13 +223,25 @@ def Trading(
 
 
 def _default_authenticator() -> Authenticator | None:
-    if defaults.PRIVATE_KEY is not None and defaults.KEYFILE is not None:
+    auth_variable_count = sum(
+        [
+            int(bool(defaults.PRIVATE_KEY)),
+            int(bool(defaults.KEYFILE)),
+            int(bool(defaults.TREZOR_PATH_OR_INDEX)),
+        ]
+    )
+    if auth_variable_count > 1:
         raise ConfigurationError(
-            "Only one of AFP_PRIVATE_KEY and AFP_KEYFILE environment "
-            "variables should be specified"
+            "Only one of AFP_PRIVATE_KEY, AFP_KEYFILE and AFP_TREZOR_PATH_OR_INDEX "
+            "environment variables should be specified"
         )
-    if defaults.PRIVATE_KEY is not None:
+
+    if defaults.PRIVATE_KEY:
         return PrivateKeyAuthenticator(defaults.PRIVATE_KEY)
-    if defaults.KEYFILE is not None:
+    if defaults.KEYFILE:
         return KeyfileAuthenticator(defaults.KEYFILE, defaults.KEYFILE_PASSWORD)
+    if defaults.TREZOR_PATH_OR_INDEX:
+        return TrezorAuthenticator(
+            defaults.TREZOR_PATH_OR_INDEX, defaults.TREZOR_PASSPHRASE
+        )
     return None

afp/auth.py

@@ -1,17 +1,35 @@
+import atexit
 import json
 import os
+import sys
 from typing import Protocol, cast
 
+import trezorlib.ethereum as trezor_eth
 from eth_account.account import Account
 from eth_account.datastructures import SignedTransaction
 from eth_account.messages import encode_defunct
 from eth_account.signers.local import LocalAccount
 from eth_account.types import TransactionDictType
+from eth_account._utils.legacy_transactions import (
+    encode_transaction,
+    serializable_unsigned_transaction_from_dict,
+)
 from eth_typing.evm import ChecksumAddress
+from eth_utils.conversions import to_int
+from eth_utils.crypto import keccak
 from hexbytes import HexBytes
+from trezorlib.client import TrezorClient
+from trezorlib.client import get_default_client  # type: ignore
+from trezorlib.ui import TrezorClientUI
+from trezorlib.tools import parse_path
+from trezorlib.transport import DeviceIsBusy
+from web3 import Web3
 from web3.constants import CHECKSUM_ADDRESSS_ZERO
 from web3.types import TxParams
 
+from .constants import TREZOR_DEFAULT_PREFIX
+from .exceptions import DeviceError
+
 
 class Authenticator(Protocol):
     address: ChecksumAddress
@@ -69,12 +87,154 @@ class KeyfileAuthenticator(PrivateKeyAuthenticator):
     key_file : str
         The path to the keyfile.
     password : str
-        The password for decrypting the keyfile.
+        The password for decrypting the keyfile. Defaults to no password.
     """
 
-    def __init__(self, key_file: str, password: str) -> None:
+    def __init__(self, key_file: str, password: str = "") -> None:
         with open(os.path.expanduser(key_file), encoding="utf8") as f:
             key_data = json.load(f)
 
         private_key = Account.decrypt(key_data, password=password)
         super().__init__(private_key.to_0x_hex())
+
+
+class TrezorAuthenticator(Authenticator):
+    """Authenticates with a Trezor device.
+
+    Parameters
+    ----------
+    path_or_index: str or int
+        The full derivation path of the account, e.g. `m/44'/60'/0'/0/123`; or the
+        index of the account at the default Trezor derivation prefix for Ethereum
+        accounts `m/44'/60'/0'/0`, e.g. `123`.
+    passphrase: str
+        The passphrase for the Trezor device. Defaults to no passphrase.
+    """
+
+    client: TrezorClient[TrezorClientUI]
+
+    def __init__(self, path_or_index: str | int, passphrase: str = ""):
+        if isinstance(path_or_index, int) or path_or_index.isdigit():
+            path_str = f"{TREZOR_DEFAULT_PREFIX}/{int(path_or_index)}"
+        else:
+            path_str = path_or_index.replace("'", "h")
+        try:
+            self.path = parse_path(path_str)
+        except ValueError as exc:
+            raise DeviceError(
+                f"Invalid Trezor BIP32 derivation path '{path_str}'"
+            ) from exc
+        self.client = self._get_client(passphrase)
+        atexit.register(self.client.end_session)
+
+        address_str = trezor_eth.get_address(  # type: ignore
+            self.client, self.path
+        )
+        self.address = Web3.to_checksum_address(address_str)
+
+    def sign_transaction(self, params: TxParams) -> SignedTransaction:
+        assert "chainId" in params
+        assert "gas" in params
+        assert "nonce" in params
+        assert "to" in params
+        assert "value" in params
+        data_bytes = HexBytes(params["data"] if "data" in params else b"")
+
+        print("[Confirm on Trezor device]", file=sys.stderr)
+
+        if "gasPrice" in params and params["gasPrice"]:
+            v_int, r_bytes, s_bytes = trezor_eth.sign_tx(  # type: ignore
+                self.client,
+                self.path,
+                nonce=cast(int, params["nonce"]),
+                gas_price=cast(int, params["gasPrice"]),
+                gas_limit=params["gas"],
+                to=cast(str, params["to"]),
+                value=cast(int, params["value"]),
+                data=data_bytes,
+                chain_id=params["chainId"],
+            )
+        else:
+            assert "maxFeePerGas" in params
+            assert "maxPriorityFeePerGas" in params
+            v_int, r_bytes, s_bytes = trezor_eth.sign_tx_eip1559(  # type: ignore
+                self.client,
+                self.path,
+                nonce=cast(int, params["nonce"]),
+                gas_limit=params["gas"],
+                to=cast(str, params["to"]),
+                value=cast(int, params["value"]),
+                data=data_bytes,
+                chain_id=params["chainId"],
+                max_gas_fee=int(params["maxFeePerGas"]),
+                max_priority_fee=int(params["maxPriorityFeePerGas"]),
+            )
+
+        r_int = to_int(r_bytes)
+        s_int = to_int(s_bytes)
+        filtered_tx = dict((k, v) for (k, v) in params.items() if k not in ("from"))
+        # In a LegacyTransaction, "type" is not a valid field. See EIP-2718.
+        if "type" in filtered_tx and filtered_tx["type"] == "0x0":
+            filtered_tx.pop("type")
+        tx_unsigned = serializable_unsigned_transaction_from_dict(
+            cast(TransactionDictType, filtered_tx)
+        )
+        tx_encoded = encode_transaction(tx_unsigned, vrs=(v_int, r_int, s_int))
+        txhash = keccak(tx_encoded)
+        return SignedTransaction(
+            raw_transaction=HexBytes(tx_encoded),
+            hash=HexBytes(txhash),
+            r=r_int,
+            s=s_int,
+            v=v_int,
+        )
+
+    def sign_message(self, message: bytes) -> HexBytes:
+        print("[Confirm on Trezor device]", file=sys.stderr)
+
+        sigdata = trezor_eth.sign_message(  # type: ignore
+            self.client,
+            self.path,
+            message.decode("utf-8"),
+        )
+        return HexBytes(sigdata.signature)
+
+    @staticmethod
+    def _get_client(
+        passphrase: str,
+    ) -> TrezorClient[TrezorClientUI]:
+        ui = _NonInteractiveTrezorUI(passphrase)
+        try:
+            return cast(
+                TrezorClient[TrezorClientUI],
+                get_default_client(ui=ui),
+            )
+        except DeviceIsBusy as exc:
+            raise DeviceError("Device in use by another process") from exc
+        except Exception as exc:
+            raise DeviceError(
+                "No Trezor device found; "
+                "check device is connected, unlocked, and detected by OS"
+            ) from exc
+
+
+class _NonInteractiveTrezorUI(TrezorClientUI):
+    """Replacement for the default TrezorClientUI of the Trezor library.
+
+    Bringing up an interactive passphrase prompt is unwanted in the SDK;
+    this implementation receives the passphrase as constructor argument.
+    """
+
+    _passphrase: str
+
+    def __init__(self, passphrase: str) -> None:
+        self._passphrase = passphrase
+
+    def button_request(self, br: object) -> None:
+        pass
+
+    def get_pin(self, code: object) -> str:
+        raise DeviceError("PIN entry on host is not supported")
+
+    def get_passphrase(self, available_on_device: bool) -> str:
+        return self._passphrase

afp/constants.py

@@ -41,6 +41,9 @@ def _int_or_none(value: str | None) -> int | None:
     "Z": 12,
 }
 
+# Trezor device constants
+TREZOR_DEFAULT_PREFIX = "m/44h/60h/0h/0"
+
 schema_cids = SimpleNamespace(
     # afp-product-schemas v0.2.0
     ORACLE_CONFIG_V020="bafyreifcec2km7hxwq6oqzjlspni2mgipetjb7pqtaewh2efislzoctboi",
@@ -86,6 +89,8 @@ def _int_or_none(value: str | None) -> int | None:
     KEYFILE=os.getenv("AFP_KEYFILE", None),
     KEYFILE_PASSWORD=os.getenv("AFP_KEYFILE_PASSWORD", ""),
     PRIVATE_KEY=os.getenv("AFP_PRIVATE_KEY", None),
+    TREZOR_PATH_OR_INDEX=os.getenv("AFP_TERZOR_PATH_OR_INDEX", None),
+    TREZOR_PASSPHRASE=os.getenv("AFP_TREZOR_PASSPHRASE", ""),
     # Venue parameters
     EXCHANGE_URL=os.getenv("AFP_EXCHANGE_URL", _current_env.EXCHANGE_URL),
     # IPFS client parameters

afp/exceptions.py

@@ -21,6 +21,10 @@ class ExchangeError(AFPException):
     pass
 
 
+class DeviceError(AFPException):
+    pass
+
+
 # Exchange error sub-types
 
 

pyproject.toml

@@ -29,6 +29,7 @@ dependencies = [
     "requests>=2.32.0",
     "rfc8785>=0.1.4",
     "siwe>=4.4.0",
+    "trezor[ethereum]>=0.13.10",
     "web3>=7.6.0",
 ]
 classifiers = [
@@ -47,7 +48,7 @@ Changes = "https://github.com/autonity/afp-sdk/blob/master/CHANGELOG.md"
 
 [dependency-groups]
 dev = [
-    "check-wheel-contents>=0.6.3",
+    "check-wheel-contents>=0.6.1,<0.6.2",
     "griffe2md>=1.2.1",
     "poethepoet>=0.33.1",
     "pydistcheck>=0.10.0",