#!/usr/bin/env python3 import sys import traceback import click import Pyro4 from trezorlib import cosi from trezorlib.client import get_default_client from trezorlib.tools import parse_path from trezorlib._internal.firmware_headers import ( parse_image, VendorHeader, BootloaderImage, FirmwareImage, ) from typing import Tuple Pyro4.config.SERIALIZER = "marshal" PORT = 5001 indexmap = { "bootloader": BootloaderImage, "vendorheader": VendorHeader, "firmware": FirmwareImage, } PATH = "10018h/{}h" TREZOR = None def make_commit(fw_or_type, digest, public_keys): path = PATH.format(fw_or_type.BIP32_INDEX) address_n = parse_path(path) # device information - show only first time click.echo( "\nUsing device {} ".format(click.style(TREZOR.features.label, bold=True)) + "at path {}".format(TREZOR.transport.get_path()) ) while True: # signing information - repeat every time click.echo("Commiting to {} hash:".format(click.style(fw_or_type.NAME, bold=True))) for partid in range(4): digest_part = digest[partid * 8 : (partid + 1) * 8] color = "red" if partid % 2 else "cyan" digest_str = click.style(digest_part.hex().upper(), fg=color) click.echo("\t" + digest_str) click.echo("Using path: {}".format(click.style(path, bold=True))) try: commit = cosi.commit(TREZOR, address_n, digest) if public_keys is not None and commit.pubkey not in public_keys: click.echo("\n\nPublic key {} is unknown.".format(commit.pubkey.hex())) if click.confirm("Retry with a different passphrase?", default=True): TREZOR.init_device() continue return commit.pubkey, commit.commitment except Exception as e: click.echo(e) traceback.print_exc() click.echo("Trying again ...\n\n") @Pyro4.expose class KeyctlProxy: def __init__( self, daemon, image_type, digest: bytes, commit: Tuple[bytes, bytes] ) -> None: self.daemon = daemon self.name = image_type.NAME self.address_n = parse_path(PATH.format(image_type.BIP32_INDEX)) self.digest = digest self.commit = commit self.signature = None self.global_params = None def _check_name_digest(self, name, digest): if name != self.name or digest != self.digest: click.echo("ERROR! Remote wants to sign {} with digest {}".format(name, digest.hex())) click.echo("Expected: {} with digest {}".format(self.name, self.digest.hex())) raise ValueError("Unexpected index/digest") def get_commit(self, name, digest): self._check_name_digest(name, digest) click.echo("Sending commitment!") return self.commit def _make_signature(self, global_R, global_pk): while True: try: click.echo("\n\n\nSigning...") signature = cosi.sign( TREZOR, self.address_n, self.digest, global_R, global_pk ) return signature.signature except Exception as e: click.echo(e) traceback.print_exc() click.echo("Trying again ...") def get_signature(self, name, digest, global_R, global_pk): self._check_name_digest(name, digest) global_params = global_R, global_pk if global_params != self.global_params: self.signature = self._make_signature(global_R, global_pk) self.global_params = global_params click.echo("Sending signature!") return self.signature @Pyro4.oneway def finish(self): click.echo("Done! \\(^o^)/") self.daemon.shutdown() @click.command() @click.option( "-l", "--listen", "ipaddr", default="0.0.0.0", help="Bind to particular ip address" ) @click.option("-t", "--image-type", type=click.Choice(indexmap.keys())) @click.option("-d", "--digest") @click.argument("fw_file", type=click.File("rb"), required=False) def cli(ipaddr, fw_file, image_type, digest): """Participate in signing of firmware. Specify either fw_file to auto-detect type and digest, or use -t and -d to specify the type and digest manually. """ global TREZOR public_keys = None if fw_file: if image_type or digest: raise click.ClickException("Do not specify fw_file together with -t/-d") image_type = parse_image(fw_file.read()) digest = image_type.digest() public_keys = image_type.public_keys click.echo(image_type.format()) if not fw_file and (not image_type or not digest): raise click.ClickException("Please specify either fw_file or -t and -d") try: TREZOR = get_default_client() TREZOR.ui.always_prompt = True except Exception as e: raise click.ClickException("Please connect a Trezor and retry.") from e pubkey, R = make_commit(image_type, digest, public_keys) daemon = Pyro4.Daemon(host=ipaddr, port=PORT) proxy = KeyctlProxy(daemon, image_type, digest, (pubkey, R)) uri = daemon.register(proxy, "keyctl") click.echo("keyctl-proxy running at URI: {}".format(uri)) click.echo("Press Ctrl+C to abort.") daemon.requestLoop() if __name__ == "__main__": cli()