bump version

This enables git-cola support TREZOR-based commit signatues,
together with "cola.signcommits=true" setting.

.pylintrc

@@ -1,2 +1,2 @@
 [MESSAGES CONTROL]
-disable=fixme, invalid-name, missing-docstring, locally-disabled, unbalanced-tuple-unpacking
+disable=invalid-name, missing-docstring, locally-disabled, unbalanced-tuple-unpacking

.travis.yml

@@ -5,9 +5,22 @@ python:
   - "3.4"
   - "3.5"
 
+cache:
+  directories:
+    - $HOME/.cache/pip
+
+addons:
+    apt:
+        packages:
+            - libudev-dev
+            - libusb-1.0-0-dev
+
+before_install:
+  - pip install -U setuptools pylint coverage pep8 pydocstyle "pip>=7.0" wheel
+  - pip install -e git+https://github.com/keepkey/python-keepkey@6e8baa8b935e830d05f87b6dfd9bc7c927a96dc3#egg=keepkey
+
 install:
-  - pip install ecdsa ed25519 semver  # test without trezorlib for now
-  - pip install -U pylint coverage pep8 pydocstyle  # use latest tools
+  - pip install -e .
 
 script:
   - pep8 trezor_agent

README-GPG.md

@@ -18,87 +18,20 @@ Install latest `trezor-agent` package from GitHub:
 $ pip install --user git+https://github.com/romanz/trezor-agent.git
 ```
 
-Define your GPG user ID as an environment variable:
-```
-$ TREZOR_GPG_USER_ID="John Doe <john@doe.bit>"
-```
+# Quickstart
 
-There are two ways to generate TREZOR-based GPG public keys, as described below.
+## Identity creation
+[![asciicast](https://asciinema.org/a/c2yodst21h9obttkn9wgf3783.png)](https://asciinema.org/a/c2yodst21h9obttkn9wgf3783)
 
-## 1. generate a new GPG identity:
-
-```
-$ trezor-gpg create "${TREZOR_GPG_USER_ID}" | gpg2 --import           # use the TREZOR to confirm signing the primary key
-gpg: key 5E4D684D: public key "John Doe <john@doe.bit>" imported
-gpg: Total number processed: 1
-gpg:               imported: 1
-
-$ gpg2 --edit "${TREZOR_GPG_USER_ID}" trust                           # set this key to ultimate trust (option #5)
-
-$ gpg2 -k
-/home/roman/.gnupg/pubring.kbx
-------------------------------
-pub   nistp256/5E4D684D 2016-06-17 [SC]
-uid         [ultimate] John Doe <john@doe.bit>
-sub   nistp256/A31D9E25 2016-06-17 [E]
-```
-
-## 2. generate a new subkey for an existing GPG identity:
-
-```
-$ gpg2 -k                                                             # suppose there is already a GPG primary key
-/home/roman/.gnupg/pubring.kbx
-------------------------------
-pub   rsa2048/87BB07B4 2016-06-17 [SC]
-uid         [ultimate] John Doe <john@doe.bit>
-sub   rsa2048/7176D31F 2016-06-17 [E]
-
-$ trezor-gpg create --subkey "${TREZOR_GPG_USER_ID}" | gpg2 --import  # use the TREZOR to confirm signing the subkey
-gpg: key 87BB07B4: "John Doe <john@doe.bit>" 2 new signatures
-gpg: key 87BB07B4: "John Doe <john@doe.bit>" 2 new subkeys
-gpg: Total number processed: 1
-gpg:            new subkeys: 2
-gpg:         new signatures: 2
-
-$ gpg2 -k
-/home/roman/.gnupg/pubring.kbx
-------------------------------
-pub   rsa2048/87BB07B4 2016-06-17 [SC]
-uid         [ultimate] John Doe <john@doe.bit>
-sub   rsa2048/7176D31F 2016-06-17 [E]
-sub   nistp256/DDE80B36 2016-06-17 [S]
-sub   nistp256/E3D0BA19 2016-06-17 [E]
-```
-
-# Usage examples:
-
-## Start the TREZOR-based gpg-agent:
-```
-$ trezor-gpg agent &
-```
-Note: this agent intercepts all GPG requests, so make sure to close it (e.g. by using `killall trezor-gpg`),
-when you are done with the TREZOR-based GPG operations.
-
-## Sign and verify GPG messages:
-```
-$ echo "Hello World!" | gpg2 --sign | gpg2 --verify
-gpg: Signature made Fri 17 Jun 2016 08:55:13 PM IDT using ECDSA key ID 5E4D684D
-gpg: Good signature from "John Doe <john@doe.bit>" [ultimate]
-```
-## Encrypt and decrypt GPG messages:
-```
-$ date | gpg2 --encrypt -r "${TREZOR_GPG_USER_ID}" | gpg2 --decrypt
-gpg: encrypted with 256-bit ECDH key, ID A31D9E25, created 2016-06-17
-      "John Doe <john@doe.bit>"
-Fri Jun 17 20:55:31 IDT 2016
-```
+## Sample usage (signature and decryption)
+[![asciicast](https://asciinema.org/a/7x0h9tyoyu5ar6jc8y9oih0ba.png)](https://asciinema.org/a/7x0h9tyoyu5ar6jc8y9oih0ba)
 
 ## Git commit & tag signatures:
 Git can use GPG to sign and verify commits and tags (see [here](https://git-scm.com/book/en/v2/Git-Tools-Signing-Your-Work)):
 ```
-$ git config --local gpg.program gpg2
+$ git config --local gpg.program $(which gpg2)
 $ git commit --gpg-sign                      # create GPG-signed commit
 $ git log --show-signature -1                # verify commit signature
-$ git tag --sign "TAG"                       # create GPG-signed tag
-$ git verify-tag "TAG"                       # verify tag signature
+$ git tag --sign "v1.2.3"                    # create GPG-signed tag
+$ git verify-tag "v1.2.3"                    # verify tag signature
 ```

scripts/gpg-init

@@ -0,0 +1,32 @@
+#!/bin/bash
+set -eu
+
+USER_ID="${1}"
+HOMEDIR=~/.gnupg/trezor
+CURVE=${CURVE:="nist256p1"}  # or "ed25519"
+TIMESTAMP=${TIMESTAMP:=`date +%s`}  # key creation timestamp
+
+# Prepare new GPG home directory for TREZOR-based identity
+rm -rf "${HOMEDIR}"
+mkdir -p "${HOMEDIR}"
+chmod 700 "${HOMEDIR}"
+
+# Generate new GPG identity and import into GPG keyring
+trezor-gpg-create -v "${USER_ID}" -t "${TIMESTAMP}" -e "${CURVE}" > "${HOMEDIR}/pubkey.asc"
+gpg2 --homedir "${HOMEDIR}" --import < "${HOMEDIR}/pubkey.asc"
+rm -f "${HOMEDIR}/S.gpg-agent"  # (otherwise, our agent won't be started automatically)
+
+# Make new GPG identity with "ultimate" trust (via its fingerprint)
+FINGERPRINT=$(gpg2 --homedir "${HOMEDIR}" --list-public-keys --with-colons | sed -n -E 's/^fpr:::::::::([0-9A-F]+):$/\1/p' | head -n1)
+echo "${FINGERPRINT}:6" | gpg2 --homedir "${HOMEDIR}" --import-ownertrust
+
+# Prepare GPG configuration file
+echo "# TREZOR-based GPG configuration
+agent-program $(which trezor-gpg-agent)
+personal-digest-preferences SHA512
+" | tee "${HOMEDIR}/gpg.conf"
+
+echo "# TREZOR-based GPG agent emulator
+log-file ${HOMEDIR}/gpg-agent.log
+verbosity 2
+" | tee "${HOMEDIR}/gpg-agent.conf"

scripts/gpg-shell

@@ -0,0 +1,16 @@
+#!/bin/bash
+set -eu
+
+export GNUPGHOME=~/.gnupg/trezor
+
+# Make sure that the device is unlocked before starting the shell
+trezor-gpg-unlock
+
+COMMAND=$*
+if [ -z "${COMMAND}" ]
+then
+	gpg2 --list-public-keys
+	${SHELL}
+else
+	${COMMAND}
+fi

setup.py

@@ -3,13 +3,14 @@ from setuptools import setup
 
 setup(
     name='trezor_agent',
-    version='0.7.3',
+    version='0.8.0',
     description='Using Trezor as hardware SSH agent',
     author='Roman Zeyde',
     author_email='roman.zeyde@gmail.com',
     url='http://github.com/romanz/trezor-agent',
-    packages=['trezor_agent', 'trezor_agent.gpg'],
-    install_requires=['ecdsa>=0.13', 'ed25519>=1.4', 'Cython>=0.23.4', 'protobuf>=3.0.0', 'trezor>=0.7.4', 'semver>=2.2'],
+    packages=['trezor_agent', 'trezor_agent.device', 'trezor_agent.gpg'],
+    install_requires=['ecdsa>=0.13', 'ed25519>=1.4', 'Cython>=0.23.4', 'protobuf>=3.0.0', 'semver>=2.2',
+                      'trezor>=0.7.6', 'keepkey>=0.7.3', 'ledgerblue>=0.1.8'],
     platforms=['POSIX'],
     classifiers=[
         'Environment :: Console',
@@ -21,19 +22,17 @@ setup(
         'Operating System :: POSIX',
         'Programming Language :: Python :: 2.7',
         'Programming Language :: Python :: 3.4',
+        'Programming Language :: Python :: 3.5',
         'Topic :: Software Development :: Libraries :: Python Modules',
         'Topic :: System :: Networking',
         'Topic :: Communications',
         'Topic :: Security',
         'Topic :: Utilities',
     ],
-    extras_require={
-        'trezorlib': ['python-trezor>=0.7.4'],
-        'keepkeylib': ['keepkey>=0.7.3'],
-    },
     entry_points={'console_scripts': [
         'trezor-agent = trezor_agent.__main__:run_agent',
-        'trezor-git = trezor_agent.__main__:run_git',
-        'trezor-gpg = trezor_agent.gpg.__main__:main',
+        'trezor-gpg-create = trezor_agent.gpg.__main__:main_create',
+        'trezor-gpg-agent = trezor_agent.gpg.__main__:main_agent',
+        'trezor-gpg-unlock = trezor_agent.gpg.__main__:auto_unlock',
     ]},
 )

trezor_agent/__main__.py

@@ -7,14 +7,14 @@ import re
 import subprocess
 import sys
 
-from . import client, formats, protocol, server, util
+from . import client, device, formats, protocol, server, util
 
 log = logging.getLogger(__name__)
 
 
 def ssh_args(label):
     """Create SSH command for connecting specified server."""
-    identity = util.string_to_identity(label, identity_type=dict)
+    identity = device.interface.string_to_identity(label)
 
     args = []
     if 'port' in identity:
@@ -93,13 +93,11 @@ def git_host(remote_name, attributes):
             return '{user}@{host}'.format(**match.groupdict())
 
 
-def run_server(conn, public_key, command, debug, timeout):
+def run_server(conn, public_keys, command, debug, timeout):
     """Common code for run_agent and run_git below."""
     try:
         signer = conn.sign_ssh_challenge
-        public_key = formats.import_public_key(public_key)
-        log.info('using SSH public key: %s', public_key['fingerprint'])
-        handler = protocol.Handler(keys=[public_key], signer=signer,
+        handler = protocol.Handler(keys=public_keys, signer=signer,
                                    debug=debug)
         with server.serve(handler=handler, timeout=timeout) as env:
             return server.run_process(command=command, environ=env)
@@ -119,58 +117,50 @@ def handle_connection_error(func):
     return wrapper
 
 
+def parse_config(fname):
+    """Parse config file into a list of Identity objects."""
+    contents = open(fname).read()
+    for identity_str, curve_name in re.findall(r'\<(.*?)\|(.*?)\>', contents):
+        yield device.interface.Identity(identity_str=identity_str,
+                                        curve_name=curve_name)
+
+
 @handle_connection_error
 def run_agent(client_factory=client.Client):
     """Run ssh-agent using given hardware client factory."""
     args = create_agent_parser().parse_args()
     util.setup_logging(verbosity=args.verbose)
 
-    with client_factory(curve=args.ecdsa_curve_name) as conn:
-        label = args.identity
-        command = args.command
+    conn = client_factory(device=device.detect())
+    if args.identity.startswith('/'):
+        identities = list(parse_config(fname=args.identity))
+    else:
+        identities = [device.interface.Identity(
+            identity_str=args.identity, curve_name=args.ecdsa_curve_name)]
+    for index, identity in enumerate(identities):
+        identity.identity_dict['proto'] = 'ssh'
+        log.info('identity #%d: %s', index, identity)
 
-        public_key = conn.get_public_key(label=label)
+    command = args.command
 
-        if args.connect:
-            command = ssh_args(label) + args.command
-            log.debug('SSH connect: %r', command)
+    public_keys = [conn.get_public_key(i) for i in identities]
 
-        use_shell = bool(args.shell)
-        if use_shell:
-            command = os.environ['SHELL']
-            log.debug('using shell: %r', command)
+    if args.connect:
+        command = ssh_args(args.identity) + args.command
+        log.debug('SSH connect: %r', command)
 
-        if not command:
-            sys.stdout.write(public_key)
-            return
+    use_shell = bool(args.shell)
+    if use_shell:
+        command = os.environ['SHELL']
+        log.debug('using shell: %r', command)
 
-        return run_server(conn=conn, public_key=public_key, command=command,
-                          debug=args.debug, timeout=args.timeout)
+    if not command:
+        for pk in public_keys:
+            sys.stdout.write(pk)
+        return
 
-
-@handle_connection_error
-def run_git(client_factory=client.Client):
-    """Run git under ssh-agent using given hardware client factory."""
-    args = create_git_parser().parse_args()
-    util.setup_logging(verbosity=args.verbose)
-
-    with client_factory(curve=args.ecdsa_curve_name) as conn:
-        label = git_host(args.remote, ['pushurl', 'url'])
-        if not label:
-            log.error('Could not find "%s" SSH remote in .git/config',
-                      args.remote)
-            return
-
-        public_key = conn.get_public_key(label=label)
-
-        if not args.test:
-            if args.command:
-                command = ['git'] + args.command
-            else:
-                sys.stdout.write(public_key)
-                return
-        else:
-            command = ['ssh', '-T', label]
-
-        return run_server(conn=conn, public_key=public_key, command=command,
-                          debug=args.debug, timeout=args.timeout)
+    public_keys = [formats.import_public_key(pk) for pk in public_keys]
+    for pk, identity in zip(public_keys, identities):
+        pk['identity'] = identity
+    return run_server(conn=conn, public_keys=public_keys, command=command,
+                      debug=args.debug, timeout=args.timeout)

trezor_agent/client.py

@@ -3,11 +3,10 @@ Connection to hardware authentication device.
 
 It is used for getting SSH public keys and ECDSA signing of server requests.
 """
-import binascii
 import io
 import logging
 
-from . import factory, formats, util
+from . import formats, util
 
 log = logging.getLogger(__name__)
 
@@ -15,79 +14,36 @@ log = logging.getLogger(__name__)
 class Client(object):
     """Client wrapper for SSH authentication device."""
 
-    def __init__(self, loader=factory.load, curve=formats.CURVE_NIST256):
+    def __init__(self, device):
         """Connect to hardware device."""
-        client_wrapper = loader()
-        self.client = client_wrapper.connection
-        self.identity_type = client_wrapper.identity_type
-        self.device_name = client_wrapper.device_name
-        self.call_exception = client_wrapper.call_exception
-        self.curve = curve
+        self.device = device
 
-    def __enter__(self):
-        """Start a session, and test connection."""
-        msg = 'Hello World!'
-        assert self.client.ping(msg) == msg
-        return self
+    def get_public_key(self, identity):
+        """Get SSH public key from the device."""
+        with self.device:
+            pubkey = self.device.pubkey(identity)
 
-    def __exit__(self, *args):
-        """Keep the session open (doesn't forget PIN)."""
-        log.info('disconnected from %s', self.device_name)
-        self.client.close()
+        vk = formats.decompress_pubkey(pubkey=pubkey,
+                                       curve_name=identity.curve_name)
+        return formats.export_public_key(vk=vk,
+                                         label=str(identity))
 
-    def get_identity(self, label, index=0):
-        """Parse label string into Identity protobuf."""
-        identity = util.string_to_identity(label, self.identity_type)
-        identity.proto = 'ssh'
-        identity.index = index
-        return identity
-
-    def get_public_key(self, label):
-        """Get SSH public key corresponding to specified by label."""
-        identity = self.get_identity(label=label)
-        label = util.identity_to_string(identity)  # canonize key label
-        log.info('getting "%s" public key (%s) from %s...',
-                 label, self.curve, self.device_name)
-        addr = util.get_bip32_address(identity)
-        node = self.client.get_public_node(n=addr,
-                                           ecdsa_curve_name=self.curve)
-
-        pubkey = node.node.public_key
-        vk = formats.decompress_pubkey(pubkey=pubkey, curve_name=self.curve)
-        return formats.export_public_key(vk=vk, label=label)
-
-    def sign_ssh_challenge(self, label, blob):
-        """Sign given blob using a private key, specified by the label."""
-        identity = self.get_identity(label=label)
+    def sign_ssh_challenge(self, blob, identity):
+        """Sign given blob using a private key on the device."""
         msg = _parse_ssh_blob(blob)
         log.debug('%s: user %r via %r (%r)',
                   msg['conn'], msg['user'], msg['auth'], msg['key_type'])
-        log.debug('nonce: %s', binascii.hexlify(msg['nonce']))
-        log.debug('fingerprint: %s', msg['public_key']['fingerprint'])
+        log.debug('nonce: %r', msg['nonce'])
+        fp = msg['public_key']['fingerprint']
+        log.debug('fingerprint: %s', fp)
         log.debug('hidden challenge size: %d bytes', len(blob))
 
         log.info('please confirm user "%s" login to "%s" using %s...',
-                 msg['user'].decode('ascii'), label, self.device_name)
+                 msg['user'].decode('ascii'), identity,
+                 self.device)
 
-        try:
-            result = self.client.sign_identity(identity=identity,
-                                               challenge_hidden=blob,
-                                               challenge_visual='',
-                                               ecdsa_curve_name=self.curve)
-        except self.call_exception as e:
-            code, msg = e.args
-            log.warning('%s error #%s: %s', self.device_name, code, msg)
-            raise IOError(msg)  # close current connection, keep server open
-
-        verifying_key = formats.decompress_pubkey(pubkey=result.public_key,
-                                                  curve_name=self.curve)
-        key_type, blob = formats.serialize_verifying_key(verifying_key)
-        assert blob == msg['public_key']['blob']
-        assert key_type == msg['key_type']
-        assert len(result.signature) == 65
-        assert result.signature[:1] == bytearray([0])
-
-        return result.signature[1:]
+        with self.device:
+            return self.device.sign(blob=blob, identity=identity)
 
 
 def _parse_ssh_blob(data):

trezor_agent/device/__init__.py

@@ -0,0 +1,27 @@
+"""Cryptographic hardware device management."""
+
+import logging
+
+from . import trezor
+from . import keepkey
+from . import ledger
+from . import interface
+
+log = logging.getLogger(__name__)
+
+DEVICE_TYPES = [
+    trezor.Trezor,
+    keepkey.KeepKey,
+    ledger.LedgerNanoS,
+]
+
+
+def detect():
+    """Detect the first available device and return it to the user."""
+    for device_type in DEVICE_TYPES:
+        try:
+            with device_type() as d:
+                return d
+        except interface.NotFoundError as e:
+            log.debug('device not found: %s', e)
+    raise IOError('No device found!')

trezor_agent/device/interface.py

@@ -0,0 +1,135 @@
+"""Device abstraction layer."""
+
+import hashlib
+import io
+import logging
+import re
+import struct
+
+from .. import formats, util
+
+log = logging.getLogger(__name__)
+
+_identity_regexp = re.compile(''.join([
+    '^'
+    r'(?:(?P<proto>.*)://)?',
+    r'(?:(?P<user>.*)@)?',
+    r'(?P<host>.*?)',
+    r'(?::(?P<port>\w*))?',
+    r'(?P<path>/.*)?',
+    '$'
+]))
+
+
+def string_to_identity(identity_str):
+    """Parse string into Identity dictionary."""
+    m = _identity_regexp.match(identity_str)
+    result = m.groupdict()
+    log.debug('parsed identity: %s', result)
+    return {k: v for k, v in result.items() if v}
+
+
+def identity_to_string(identity_dict):
+    """Dump Identity dictionary into its string representation."""
+    result = []
+    if identity_dict.get('proto'):
+        result.append(identity_dict['proto'] + '://')
+    if identity_dict.get('user'):
+        result.append(identity_dict['user'] + '@')
+    result.append(identity_dict['host'])
+    if identity_dict.get('port'):
+        result.append(':' + identity_dict['port'])
+    if identity_dict.get('path'):
+        result.append(identity_dict['path'])
+    log.debug('identity parts: %s', result)
+    return ''.join(result)
+
+
+class Error(Exception):
+    """Device-related error."""
+
+
+class NotFoundError(Error):
+    """Device could not be found."""
+
+
+class DeviceError(Error):
+    """"Error during device operation."""
+
+
+class Identity(object):
+    """Represent SLIP-0013 identity, together with a elliptic curve choice."""
+
+    def __init__(self, identity_str, curve_name):
+        """Configure for specific identity and elliptic curve usage."""
+        self.identity_dict = string_to_identity(identity_str)
+        self.curve_name = curve_name
+
+    def items(self):
+        """Return a copy of identity_dict items."""
+        return self.identity_dict.items()
+
+    def __str__(self):
+        """Return identity serialized to string."""
+        return '<{}|{}>'.format(identity_to_string(self.identity_dict), self.curve_name)
+
+    def get_bip32_address(self, ecdh=False):
+        """Compute BIP32 derivation address according to SLIP-0013/0017."""
+        index = struct.pack('<L', self.identity_dict.get('index', 0))
+        addr = index + identity_to_string(self.identity_dict).encode('ascii')
+        log.debug('bip32 address string: %r', addr)
+        digest = hashlib.sha256(addr).digest()
+        s = io.BytesIO(bytearray(digest))
+
+        hardened = 0x80000000
+        addr_0 = [13, 17][bool(ecdh)]
+        address_n = [addr_0] + list(util.recv(s, '<LLLL'))
+        return [(hardened | value) for value in address_n]
+
+    def get_curve_name(self, ecdh=False):
+        """Return correct curve name for device operations."""
+        if ecdh:
+            return formats.get_ecdh_curve_name(self.curve_name)
+        else:
+            return self.curve_name
+
+
+class Device(object):
+    """Abstract cryptographic hardware device interface."""
+
+    def __init__(self):
+        """C-tor."""
+        self.conn = None
+
+    def connect(self):
+        """Connect to device, otherwise raise NotFoundError."""
+        raise NotImplementedError()
+
+    def __enter__(self):
+        """Allow usage as context manager."""
+        self.conn = self.connect()
+        return self
+
+    def __exit__(self, *args):
+        """Close and mark as disconnected."""
+        try:
+            self.conn.close()
+        except Exception as e:  # pylint: disable=broad-except
+            log.exception('close failed: %s', e)
+        self.conn = None
+
+    def pubkey(self, identity, ecdh=False):
+        """Get public key (as bytes)."""
+        raise NotImplementedError()
+
+    def sign(self, identity, blob):
+        """Sign given blob and return the signature (as bytes)."""
+        raise NotImplementedError()
+
+    def ecdh(self, identity, pubkey):
+        """Get shared session key using Elliptic Curve Diffie-Hellman."""
+        raise NotImplementedError()
+
+    def __str__(self):
+        """Human-readable representation."""
+        return '{}'.format(self.__class__.__name__)

trezor_agent/device/keepkey.py

@@ -0,0 +1,34 @@
+"""KeepKey-related code (see https://www.keepkey.com/)."""
+
+from . import trezor
+from .. import formats
+
+
+def _verify_support(identity, ecdh):
+    """Make sure the device supports given configuration."""
+    protocol = identity.identity_dict['proto']
+    if protocol not in {'ssh'}:
+        raise NotImplementedError(
+            'Unsupported protocol: {}'.format(protocol))
+    if ecdh:
+        raise NotImplementedError('No support for ECDH')
+    if identity.curve_name not in {formats.CURVE_NIST256}:
+        raise NotImplementedError(
+            'Unsupported elliptic curve: {}'.format(identity.curve_name))
+
+
+class KeepKey(trezor.Trezor):
+    """Connection to KeepKey device."""
+
+    from . import keepkey_defs as defs
+
+    required_version = '>=1.0.4'
+
+    def pubkey(self, identity, ecdh=False):
+        """Return public key."""
+        _verify_support(identity, ecdh)
+        return trezor.Trezor.pubkey(self, identity=identity, ecdh=ecdh)
+
+    def ecdh(self, identity, pubkey):
+        """No support for ECDH in KeepKey firmware."""
+        _verify_support(identity, ecdh=True)

trezor_agent/device/keepkey_defs.py

@@ -0,0 +1,8 @@
+"""KeepKey-related definitions."""
+
+# pylint: disable=unused-import
+from keepkeylib.client import KeepKeyClient as Client
+from keepkeylib.client import CallException
+from keepkeylib.transport_hid import HidTransport
+from keepkeylib.messages_pb2 import PassphraseAck
+from keepkeylib.types_pb2 import IdentityType

trezor_agent/device/ledger.py

@@ -0,0 +1,111 @@
+"""Ledger-related code (see https://www.ledgerwallet.com/)."""
+
+import binascii
+import logging
+import struct
+
+from ledgerblue import comm
+
+from . import interface
+
+log = logging.getLogger(__name__)
+
+
+def _expand_path(path):
+    """Convert BIP32 path into bytes."""
+    return b''.join((struct.pack('>I', e) for e in path))
+
+
+def _convert_public_key(ecdsa_curve_name, result):
+    """Convert Ledger reply into PublicKey object."""
+    if ecdsa_curve_name == 'nist256p1':
+        if (result[64] & 1) != 0:
+            result = bytearray([0x03]) + result[1:33]
+        else:
+            result = bytearray([0x02]) + result[1:33]
+    else:
+        result = result[1:]
+        keyX = bytearray(result[0:32])
+        keyY = bytearray(result[32:][::-1])
+        if (keyX[31] & 1) != 0:
+            keyY[31] |= 0x80
+        result = b'\x00' + bytes(keyY)
+    return bytes(result)
+
+
+class LedgerNanoS(interface.Device):
+    """Connection to Ledger Nano S device."""
+
+    def connect(self):
+        """Enumerate and connect to the first USB HID interface."""
+        try:
+            return comm.getDongle()
+        except comm.CommException as e:
+            raise interface.NotFoundError(
+                '{} not connected: "{}"'.format(self, e))
+
+    def pubkey(self, identity, ecdh=False):
+        """Get PublicKey object for specified BIP32 address and elliptic curve."""
+        curve_name = identity.get_curve_name(ecdh)
+        path = _expand_path(identity.get_bip32_address(ecdh))
+        if curve_name == 'nist256p1':
+            p2 = '01'
+        else:
+            p2 = '02'
+        apdu = '800200' + p2
+        apdu = binascii.unhexlify(apdu)
+        apdu += bytearray([len(path) + 1, len(path) // 4])
+        apdu += path
+        result = bytearray(self.conn.exchange(bytes(apdu)))[1:]
+        return _convert_public_key(curve_name, result)
+
+    def sign(self, identity, blob):
+        """Sign given blob and return the signature (as bytes)."""
+        path = _expand_path(identity.get_bip32_address(ecdh=False))
+        if identity.identity_dict['proto'] == 'ssh':
+            ins = '04'
+            p1 = '00'
+        else:
+            ins = '08'
+            p1 = '00'
+        if identity.curve_name == 'nist256p1':
+            p2 = '81' if identity.identity_dict['proto'] == 'ssh' else '01'
+        else:
+            p2 = '82' if identity.identity_dict['proto'] == 'ssh' else '02'
+        apdu = '80' + ins + p1 + p2
+        apdu = binascii.unhexlify(apdu)
+        apdu += bytearray([len(blob) + len(path) + 1])
+        apdu += bytearray([len(path) // 4]) + path
+        apdu += blob
+        result = bytearray(self.conn.exchange(bytes(apdu)))
+        if identity.curve_name == 'nist256p1':
+            offset = 3
+            length = result[offset]
+            r = result[offset+1:offset+1+length]
+            if r[0] == 0:
+                r = r[1:]
+            offset = offset + 1 + length + 1
+            length = result[offset]
+            s = result[offset+1:offset+1+length]
+            if s[0] == 0:
+                s = s[1:]
+            offset = offset + 1 + length
+            return bytes(r) + bytes(s)
+        else:
+            return bytes(result[:64])
+
+    def ecdh(self, identity, pubkey):
+        """Get shared session key using Elliptic Curve Diffie-Hellman."""
+        path = _expand_path(identity.get_bip32_address(ecdh=True))
+        if identity.curve_name == 'nist256p1':
+            p2 = '01'
+        else:
+            p2 = '02'
+        apdu = '800a00' + p2
+        apdu = binascii.unhexlify(apdu)
+        apdu += bytearray([len(pubkey) + len(path) + 1])
+        apdu += bytearray([len(path) // 4]) + path
+        apdu += pubkey
+        result = bytearray(self.conn.exchange(bytes(apdu)))
+        assert result[0] == 0x04
+        return bytes(result)

trezor_agent/device/trezor.py

@@ -0,0 +1,105 @@
+"""TREZOR-related code (see http://bitcointrezor.com/)."""
+
+import binascii
+import logging
+import semver
+
+from . import interface
+
+log = logging.getLogger(__name__)
+
+
+class Trezor(interface.Device):
+    """Connection to TREZOR device."""
+
+    from . import trezor_defs as defs
+
+    required_version = '>=1.4.0'
+
+    def connect(self):
+        """Enumerate and connect to the first USB HID interface."""
+        def empty_passphrase_handler(_):
+            return self.defs.PassphraseAck(passphrase='')
+
+        for d in self.defs.HidTransport.enumerate():
+            log.debug('endpoint: %s', d)
+            transport = self.defs.HidTransport(d)
+            connection = self.defs.Client(transport)
+            connection.callback_PassphraseRequest = empty_passphrase_handler
+            f = connection.features
+            log.debug('connected to %s %s', self, f.device_id)
+            log.debug('label    : %s', f.label)
+            log.debug('vendor   : %s', f.vendor)
+            current_version = '{}.{}.{}'.format(f.major_version,
+                                                f.minor_version,
+                                                f.patch_version)
+            log.debug('version  : %s', current_version)
+            log.debug('revision : %s', binascii.hexlify(f.revision))
+            if not semver.match(current_version, self.required_version):
+                fmt = ('Please upgrade your {} firmware to {} version'
+                       ' (current: {})')
+                raise ValueError(fmt.format(self, self.required_version,
+                                            current_version))
+            connection.ping(msg='', pin_protection=True)  # unlock PIN
+            return connection
+        raise interface.NotFoundError('{} not connected'.format(self))
+
+    def close(self):
+        """Close connection."""
+        self.conn.close()
+
+    def pubkey(self, identity, ecdh=False):
+        """Return public key."""
+        curve_name = identity.get_curve_name(ecdh=ecdh)
+        log.debug('"%s" getting public key (%s) from %s',
+                  identity, curve_name, self)
+        addr = identity.get_bip32_address(ecdh=ecdh)
+        result = self.conn.get_public_node(n=addr,
+                                           ecdsa_curve_name=curve_name)
+        log.debug('result: %s', result)
+        return result.node.public_key
+
+    def _identity_proto(self, identity):
+        result = self.defs.IdentityType()
+        for name, value in identity.items():
+            setattr(result, name, value)
+        return result
+
+    def sign(self, identity, blob):
+        """Sign given blob and return the signature (as bytes)."""
+        curve_name = identity.get_curve_name(ecdh=False)
+        log.debug('"%s" signing %r (%s) on %s',
+                  identity, blob, curve_name, self)
+        try:
+            result = self.conn.sign_identity(
+                identity=self._identity_proto(identity),
+                challenge_hidden=blob,
+                challenge_visual='',
+                ecdsa_curve_name=curve_name)
+            log.debug('result: %s', result)
+            assert len(result.signature) == 65
+            assert result.signature[:1] == b'\x00'
+            return result.signature[1:]
+        except self.defs.CallException as e:
+            msg = '{} error: {}'.format(self, e)
+            log.debug(msg, exc_info=True)
+            raise interface.DeviceError(msg)
+
+    def ecdh(self, identity, pubkey):
+        """Get shared session key using Elliptic Curve Diffie-Hellman."""
+        curve_name = identity.get_curve_name(ecdh=True)
+        log.debug('"%s" shared session key (%s) for %r from %s',
+                  identity, curve_name, pubkey, self)
+        try:
+            result = self.conn.get_ecdh_session_key(
+                identity=self._identity_proto(identity),
+                peer_public_key=pubkey,
+                ecdsa_curve_name=curve_name)
+            log.debug('result: %s', result)
+            assert len(result.session_key) in {65, 33}  # NIST256 or Curve25519
+            assert result.session_key[:1] == b'\x04'
+            return result.session_key
+        except self.defs.CallException as e:
+            msg = '{} error: {}'.format(self, e)
+            log.debug(msg, exc_info=True)
+            raise interface.DeviceError(msg)

trezor_agent/device/trezor_defs.py

@@ -0,0 +1,8 @@
+"""TREZOR-related definitions."""
+
+# pylint: disable=unused-import
+from trezorlib.client import TrezorClient as Client
+from trezorlib.client import CallException
+from trezorlib.transport_hid import HidTransport
+from trezorlib.messages_pb2 import PassphraseAck
+from trezorlib.types_pb2 import IdentityType

trezor_agent/factory.py

@@ -1,255 +0,0 @@
-"""Thin wrapper around trezor/keepkey libraries."""
-from __future__ import absolute_import
-
-import binascii
-import collections
-import logging
-
-import semver
-
-from . import util
-
-log = logging.getLogger(__name__)
-
-ClientWrapper = collections.namedtuple(
-    'ClientWrapper',
-    ['connection', 'identity_type', 'device_name', 'call_exception'])
-
-
-# pylint: disable=too-many-arguments
-def _load_client(name, client_type, hid_transport,
-                 passphrase_ack, identity_type,
-                 required_version, call_exception):
-
-    def empty_passphrase_handler(_):
-        return passphrase_ack(passphrase='')
-
-    for d in hid_transport.enumerate():
-        connection = client_type(hid_transport(d))
-        connection.callback_PassphraseRequest = empty_passphrase_handler
-        f = connection.features
-        log.debug('connected to %s %s', name, f.device_id)
-        log.debug('label    : %s', f.label)
-        log.debug('vendor   : %s', f.vendor)
-        current_version = '{}.{}.{}'.format(f.major_version,
-                                            f.minor_version,
-                                            f.patch_version)
-        log.debug('version  : %s', current_version)
-        log.debug('revision : %s', binascii.hexlify(f.revision))
-        if not semver.match(current_version, required_version):
-            fmt = 'Please upgrade your {} firmware to {} version (current: {})'
-            raise ValueError(fmt.format(name,
-                                        required_version,
-                                        current_version))
-        yield ClientWrapper(connection=connection,
-                            identity_type=identity_type,
-                            device_name=name,
-                            call_exception=call_exception)
-        return
-
-
-def _load_trezor():
-    try:
-        from trezorlib.client import TrezorClient, CallException
-        from trezorlib.transport_hid import HidTransport
-        from trezorlib.messages_pb2 import PassphraseAck
-        from trezorlib.types_pb2 import IdentityType
-        return _load_client(name='Trezor',
-                            client_type=TrezorClient,
-                            hid_transport=HidTransport,
-                            passphrase_ack=PassphraseAck,
-                            identity_type=IdentityType,
-                            required_version='>=1.4.0',
-                            call_exception=CallException)
-    except ImportError as e:
-        log.warning('%s: install via "pip install trezor" '
-                    'if you need to support this device', e)
-
-
-def _load_keepkey():
-    try:
-        from keepkeylib.client import KeepKeyClient, CallException
-        from keepkeylib.transport_hid import HidTransport
-        from keepkeylib.messages_pb2 import PassphraseAck
-        from keepkeylib.types_pb2 import IdentityType
-        return _load_client(name='KeepKey',
-                            client_type=KeepKeyClient,
-                            hid_transport=HidTransport,
-                            passphrase_ack=PassphraseAck,
-                            identity_type=IdentityType,
-                            required_version='>=1.0.4',
-                            call_exception=CallException)
-    except ImportError as e:
-        log.warning('%s: install via "pip install keepkey" '
-                    'if you need to support this device', e)
-
-
-def _load_ledger():
-    import struct
-
-    class LedgerClientConnection(object):
-        def __init__(self, dongle):
-            self.dongle = dongle
-
-        @staticmethod
-        def expand_path(path):
-            result = ""
-            for pathElement in path:
-                result = result + struct.pack(">I", pathElement)
-            return result
-
-        @staticmethod
-        def convert_public_key(ecdsa_curve_name, result):
-            from trezorlib.messages_pb2 import PublicKey  # pylint: disable=import-error
-            if ecdsa_curve_name == "nist256p1":
-                if (result[64] & 1) != 0:
-                    result = bytearray([0x03]) + result[1:33]
-                else:
-                    result = bytearray([0x02]) + result[1:33]
-            else:
-                result = result[1:]
-                keyX = bytearray(result[0:32])
-                keyY = bytearray(result[32:][::-1])
-                if (keyX[31] & 1) != 0:
-                    keyY[31] |= 0x80
-                result = chr(0) + str(keyY)
-            publicKey = PublicKey()
-            publicKey.node.public_key = str(result)
-            return publicKey
-
-        # pylint: disable=unused-argument
-        def get_public_node(self, n, ecdsa_curve_name="secp256k1", show_display=False):
-            donglePath = LedgerClientConnection.expand_path(n)
-            if ecdsa_curve_name == "nist256p1":
-                p2 = "01"
-            else:
-                p2 = "02"
-            apdu = "800200" + p2
-            apdu = apdu.decode('hex')
-            apdu += chr(len(donglePath) + 1) + chr(len(donglePath) / 4)
-            apdu += donglePath
-            result = bytearray(self.dongle.exchange(bytes(apdu)))[1:]
-            return LedgerClientConnection.convert_public_key(ecdsa_curve_name, result)
-
-        # pylint: disable=too-many-locals
-        def sign_identity(self, identity, challenge_hidden, challenge_visual,
-                          ecdsa_curve_name="secp256k1"):
-            from trezorlib.messages_pb2 import SignedIdentity  # pylint: disable=import-error
-            n = util.get_bip32_address(identity)
-            donglePath = LedgerClientConnection.expand_path(n)
-            if identity.proto == 'ssh':
-                ins = "04"
-                p1 = "00"
-            else:
-                ins = "08"
-                p1 = "00"
-            if ecdsa_curve_name == "nist256p1":
-                p2 = "81" if identity.proto == 'ssh' else "01"
-            else:
-                p2 = "82" if identity.proto == 'ssh' else "02"
-            apdu = "80" + ins + p1 + p2
-            apdu = apdu.decode('hex')
-            apdu += chr(len(challenge_hidden) + len(donglePath) + 1)
-            apdu += chr(len(donglePath) / 4) + donglePath
-            apdu += challenge_hidden
-            result = bytearray(self.dongle.exchange(bytes(apdu)))
-            if ecdsa_curve_name == "nist256p1":
-                offset = 3
-                length = result[offset]
-                r = result[offset+1:offset+1+length]
-                if r[0] == 0:
-                    r = r[1:]
-                offset = offset + 1 + length + 1
-                length = result[offset]
-                s = result[offset+1:offset+1+length]
-                if s[0] == 0:
-                    s = s[1:]
-                offset = offset + 1 + length
-                signature = SignedIdentity()
-                signature.signature = chr(0) + str(r) + str(s)
-                if identity.proto == 'ssh':
-                    keyData = result[offset:]
-                    pk = LedgerClientConnection.convert_public_key(ecdsa_curve_name, keyData)
-                    signature.public_key = pk.node.public_key
-                return signature
-            else:
-                signature = SignedIdentity()
-                signature.signature = chr(0) + str(result[0:64])
-                if identity.proto == 'ssh':
-                    keyData = result[64:]
-                    pk = LedgerClientConnection.convert_public_key(ecdsa_curve_name, keyData)
-                    signature.public_key = pk.node.public_key
-                return signature
-
-        def get_ecdh_session_key(self, identity, peer_public_key, ecdsa_curve_name="secp256k1"):
-            from trezorlib.messages_pb2 import ECDHSessionKey  # pylint: disable=import-error
-            n = util.get_bip32_address(identity, True)
-            donglePath = LedgerClientConnection.expand_path(n)
-            if ecdsa_curve_name == "nist256p1":
-                p2 = "01"
-            else:
-                p2 = "02"
-            apdu = "800a00" + p2
-            apdu = apdu.decode('hex')
-            apdu += chr(len(peer_public_key) + len(donglePath) + 1)
-            apdu += chr(len(donglePath) / 4) + donglePath
-            apdu += peer_public_key
-            result = bytearray(self.dongle.exchange(bytes(apdu)))
-            sessionKey = ECDHSessionKey()
-            sessionKey.session_key = str(result)
-            return sessionKey
-
-        def clear_session(self):
-            pass
-
-        def close(self):
-            self.dongle.close()
-
-        # pylint: disable=unused-argument
-        # pylint: disable=no-self-use
-        def ping(self, msg, button_protection=False, pin_protection=False,
-                 passphrase_protection=False):
-            return msg
-
-    class CallException(Exception):
-        def __init__(self, code, message):
-            super(CallException, self).__init__()
-            self.args = [code, message]
-    try:
-        from ledgerblue.comm import getDongle
-    except ImportError as e:
-        log.warning('%s: install via "pip install ledgerblue" '
-                    'if you need to support this device', e)
-        return
-    # pylint: disable=bare-except
-    try:
-        from trezorlib.types_pb2 import IdentityType  # pylint: disable=import-error
-        dongle = getDongle()
-    except:
-        return
-    yield ClientWrapper(connection=LedgerClientConnection(dongle),
-                        identity_type=IdentityType,
-                        device_name="ledger",
-                        call_exception=CallException)
-
-LOADERS = [
-    _load_trezor,
-    _load_keepkey,
-    _load_ledger
-]
-
-
-def load(loaders=None):
-    """Load a single device, via specified loaders' list."""
-    loaders = loaders if loaders is not None else LOADERS
-    device_list = []
-    for loader in loaders:
-        device = loader()
-        if device:
-            device_list.extend(device)
-
-    if len(device_list) == 1:
-        return device_list[0]
-
-    msg = '{:d} devices found'.format(len(device_list))
-    raise IOError(msg)

trezor_agent/gpg/__main__.py

@@ -2,30 +2,40 @@
 """Create signatures and export public keys for GPG using TREZOR."""
 import argparse
 import contextlib
+import io
 import logging
+import os
 import sys
 import time
 
 import semver
 
-from . import agent, device, encode, keyring, protocol
-from .. import formats, server, util
+from . import agent, decode, client, encode, keyring, protocol
+from .. import device, formats, server, util
 
 log = logging.getLogger(__name__)
 
 
+def key_exists(user_id):
+    """Return True iff there is a GPG key with specified user ID."""
+    for p in decode.parse_packets(io.BytesIO(keyring.export_public_keys())):
+        if p['type'] == 'user_id' and p['value'] == user_id:
+            return True
+    return False
+
+
 def run_create(args):
     """Generate a new pubkey for a new/existing GPG identity."""
     log.warning('NOTE: in order to re-generate the exact same GPG key later, '
                 'run this command with "--time=%d" commandline flag (to set '
                 'the timestamp of the GPG key manually).', args.time)
-    conn = device.HardwareSigner(user_id=args.user_id,
-                                 curve_name=args.ecdsa_curve)
-    verifying_key = conn.pubkey(ecdh=False)
-    decryption_key = conn.pubkey(ecdh=True)
+    d = client.Client(user_id=args.user_id, curve_name=args.ecdsa_curve)
+    verifying_key = d.pubkey(ecdh=False)
+    decryption_key = d.pubkey(ecdh=True)
 
-    if args.subkey:
-        primary_bytes = keyring.export_public_key(user_id=args.user_id)
+    if key_exists(args.user_id):  # add as subkey
+        log.info('adding %s GPG subkey for "%s" to existing key',
+                 args.ecdsa_curve, args.user_id)
         # subkey for signing
         signing_key = protocol.PublicKey(
             curve_name=args.ecdsa_curve, created=args.time,
@@ -34,13 +44,16 @@ def run_create(args):
         encryption_key = protocol.PublicKey(
             curve_name=formats.get_ecdh_curve_name(args.ecdsa_curve),
             created=args.time, verifying_key=decryption_key, ecdh=True)
+        primary_bytes = keyring.export_public_key(args.user_id)
         result = encode.create_subkey(primary_bytes=primary_bytes,
                                       subkey=signing_key,
-                                      signer_func=conn.sign)
+                                      signer_func=d.sign)
         result = encode.create_subkey(primary_bytes=result,
                                       subkey=encryption_key,
-                                      signer_func=conn.sign)
-    else:
+                                      signer_func=d.sign)
+    else:  # add as primary
+        log.info('creating new %s GPG primary key for "%s"',
+                 args.ecdsa_curve, args.user_id)
         # primary key for signing
         primary = protocol.PublicKey(
             curve_name=args.ecdsa_curve, created=args.time,
@@ -52,43 +65,21 @@ def run_create(args):
 
         result = encode.create_primary(user_id=args.user_id,
                                        pubkey=primary,
-                                       signer_func=conn.sign)
+                                       signer_func=d.sign)
         result = encode.create_subkey(primary_bytes=result,
                                       subkey=subkey,
-                                      signer_func=conn.sign)
+                                      signer_func=d.sign)
 
     sys.stdout.write(protocol.armor(result, 'PUBLIC KEY BLOCK'))
 
 
-def run_agent(args):  # pylint: disable=unused-argument
-    """Run a simple GPG-agent server."""
-    sock_path = keyring.get_agent_sock_path()
-    with server.unix_domain_socket_server(sock_path) as sock:
-        for conn in agent.yield_connections(sock):
-            with contextlib.closing(conn):
-                try:
-                    agent.handle_connection(conn)
-                except Exception as e:  # pylint: disable=broad-except
-                    log.exception('gpg-agent failed: %s', e)
-
-
-def main():
-    """Main function."""
+def main_create():
+    """Main function for GPG identity creation."""
     p = argparse.ArgumentParser()
+    p.add_argument('user_id')
+    p.add_argument('-e', '--ecdsa-curve', default='nist256p1')
+    p.add_argument('-t', '--time', type=int, default=int(time.time()))
     p.add_argument('-v', '--verbose', default=0, action='count')
-    subparsers = p.add_subparsers()
-    subparsers.required = True
-    subparsers.dest = 'command'
-
-    create_cmd = subparsers.add_parser('create')
-    create_cmd.add_argument('user_id')
-    create_cmd.add_argument('-s', '--subkey', action='store_true', default=False)
-    create_cmd.add_argument('-e', '--ecdsa-curve', default='nist256p1')
-    create_cmd.add_argument('-t', '--time', type=int, default=int(time.time()))
-    create_cmd.set_defaults(run=run_create)
-
-    agent_cmd = subparsers.add_parser('agent')
-    agent_cmd.set_defaults(run=run_agent)
 
     args = p.parse_args()
     util.setup_logging(verbosity=args.verbose)
@@ -99,11 +90,38 @@ def main():
     existing_gpg = keyring.gpg_version().decode('ascii')
     required_gpg = '>=2.1.15'
     if semver.match(existing_gpg, required_gpg):
-        args.run(args)
+        run_create(args)
     else:
         log.error('Existing gpg2 has version "%s" (%s required)',
                   existing_gpg, required_gpg)
 
 
-if __name__ == '__main__':
-    main()
+def main_agent():
+    """Run a simple GPG-agent server."""
+    home_dir = os.environ.get('GNUPGHOME', os.path.expanduser('~/.gnupg/trezor'))
+    config_file = os.path.join(home_dir, 'gpg-agent.conf')
+    lines = (line.strip() for line in open(config_file))
+    lines = (line for line in lines if line and not line.startswith('#'))
+    config = dict(line.split(' ', 1) for line in lines)
+
+    util.setup_logging(verbosity=int(config['verbosity']),
+                       filename=config['log-file'])
+    sock_path = keyring.get_agent_sock_path()
+    with server.unix_domain_socket_server(sock_path) as sock:
+        for conn in agent.yield_connections(sock):
+            with contextlib.closing(conn):
+                try:
+                    agent.handle_connection(conn)
+                except Exception as e:  # pylint: disable=broad-except
+                    log.exception('gpg-agent failed: %s', e)
+
+
+def auto_unlock():
+    """Automatically unlock first found device (used for `gpg-shell`)."""
+    p = argparse.ArgumentParser()
+    p.add_argument('-v', '--verbose', default=0, action='count')
+
+    args = p.parse_args()
+    util.setup_logging(verbosity=args.verbose)
+    d = device.detect()
+    log.info('unlocked %s device', d)

trezor_agent/gpg/agent.py

@@ -1,9 +1,8 @@
 """GPG-agent utilities."""
 import binascii
-import contextlib
 import logging
 
-from . import decode, device, keyring, protocol
+from . import decode, client, keyring, protocol
 from .. import util
 
 log = logging.getLogger(__name__)
@@ -25,7 +24,7 @@ def yield_connections(sock):
 def serialize(data):
     """Serialize data according to ASSUAN protocol."""
     for c in [b'%', b'\n', b'\r']:
-        escaped = '%{:02X}'.format(c[0]).encode('ascii')
+        escaped = '%{:02X}'.format(ord(c)).encode('ascii')
         data = data.replace(c, escaped)
     return data
 
@@ -37,7 +36,6 @@ def sig_encode(r, s):
     return b'(7:sig-val(5:ecdsa(1:r32:' + r + b')(1:s32:' + s + b')))'
 
 
-@contextlib.contextmanager
 def open_connection(keygrip_bytes):
     """
     Connect to the device for the specified keygrip.
@@ -49,29 +47,28 @@ def open_connection(keygrip_bytes):
         pubkey_bytes=keyring.export_public_keys(),
         keygrip=keygrip_bytes)
     # We assume the first user ID is used to generate TREZOR-based GPG keys.
-    user_id = user_ids[0]['value']
+    user_id = user_ids[0]['value'].decode('ascii')
     curve_name = protocol.get_curve_name_by_oid(pubkey_dict['curve_oid'])
     ecdh = (pubkey_dict['algo'] == protocol.ECDH_ALGO_ID)
 
-    conn = device.HardwareSigner(user_id, curve_name=curve_name)
-    with contextlib.closing(conn):
-        pubkey = protocol.PublicKey(
-            curve_name=curve_name, created=pubkey_dict['created'],
-            verifying_key=conn.pubkey(ecdh=ecdh), ecdh=ecdh)
-        assert pubkey.key_id() == pubkey_dict['key_id']
-        assert pubkey.keygrip == keygrip_bytes
-        yield conn
+    conn = client.Client(user_id, curve_name=curve_name)
+    pubkey = protocol.PublicKey(
+        curve_name=curve_name, created=pubkey_dict['created'],
+        verifying_key=conn.pubkey(ecdh=ecdh), ecdh=ecdh)
+    assert pubkey.key_id() == pubkey_dict['key_id']
+    assert pubkey.keygrip == keygrip_bytes
+    return conn
 
 
 def pksign(keygrip, digest, algo):
     """Sign a message digest using a private EC key."""
-    assert algo == b'8', 'Unsupported hash algorithm ID {}'.format(algo)
+    log.debug('signing %r digest (algo #%s)', digest, algo)
     keygrip_bytes = binascii.unhexlify(keygrip)
-    with open_connection(keygrip_bytes) as conn:
-        r, s = conn.sign(binascii.unhexlify(digest))
-        result = sig_encode(r, s)
-        log.debug('result: %r', result)
-        return result
+    conn = open_connection(keygrip_bytes)
+    r, s = conn.sign(binascii.unhexlify(digest))
+    result = sig_encode(r, s)
+    log.debug('result: %r', result)
+    return result
 
 
 def _serialize_point(data):
@@ -105,8 +102,8 @@ def pkdecrypt(keygrip, conn):
     remote_pubkey = parse_ecdh(line)
 
     keygrip_bytes = binascii.unhexlify(keygrip)
-    with open_connection(keygrip_bytes) as conn:
-        return _serialize_point(conn.ecdh(remote_pubkey))
+    conn = open_connection(keygrip_bytes)
+    return _serialize_point(conn.ecdh(remote_pubkey))
 
 
 def handle_connection(conn):

trezor_agent/gpg/client.py

@@ -0,0 +1,44 @@
+"""Device abstraction layer for GPG operations."""
+
+import logging
+
+from .. import device, formats, util
+
+log = logging.getLogger(__name__)
+
+
+class Client(object):
+    """Sign messages and get public keys from a hardware device."""
+
+    def __init__(self, user_id, curve_name):
+        """Connect to the device and retrieve required public key."""
+        self.device = device.detect()
+        self.user_id = user_id
+        self.identity = device.interface.Identity(
+            identity_str='gpg://', curve_name=curve_name)
+        self.identity.identity_dict['host'] = user_id
+
+    def pubkey(self, ecdh=False):
+        """Return public key as VerifyingKey object."""
+        with self.device:
+            pubkey = self.device.pubkey(ecdh=ecdh, identity=self.identity)
+        return formats.decompress_pubkey(
+            pubkey=pubkey, curve_name=self.identity.curve_name)
+
+    def sign(self, digest):
+        """Sign the digest and return a serialized signature."""
+        log.info('please confirm GPG signature on %s for "%s"...',
+                 self.device, self.user_id)
+        if self.identity.curve_name == formats.CURVE_NIST256:
+            digest = digest[:32]  # sign the first 256 bits
+        log.debug('signing digest: %s', util.hexlify(digest))
+        with self.device:
+            sig = self.device.sign(blob=digest, identity=self.identity)
+        return (util.bytes2num(sig[:32]), util.bytes2num(sig[32:]))
+
+    def ecdh(self, pubkey):
+        """Derive shared secret using ECDH from remote public key."""
+        log.info('please confirm GPG decryption on %s for "%s"...',
+                 self.device, self.user_id)
+        with self.device:
+            return self.device.ecdh(pubkey=pubkey, identity=self.identity)

trezor_agent/gpg/decode.py

@@ -127,7 +127,9 @@ def _parse_signature(stream):
         log.debug('embedded sigs: %s', embedded)
         p['embedded'] = embedded
 
-    p['_is_custom'] = (protocol.CUSTOM_SUBPACKET in p['unhashed_subpackets'])
+    # Detect our custom public keys by matching subpacket data
+    p['_is_custom'] = any(protocol.CUSTOM_KEY_LABEL == subpacket[1:]
+                          for subpacket in p['unhashed_subpackets'])
 
     p['hash_prefix'] = stream.readfmt('2s')
     if p['pubkey_alg'] in ECDSA_ALGO_IDS:

trezor_agent/gpg/device.py

@@ -1,54 +0,0 @@
-"""Device abstraction layer for GPG operations."""
-
-from .. import factory, formats, util
-
-
-class HardwareSigner(object):
-    """Sign messages and get public keys from a hardware device."""
-
-    def __init__(self, user_id, curve_name):
-        """Connect to the device and retrieve required public key."""
-        self.client_wrapper = factory.load()
-        self.identity = self.client_wrapper.identity_type()
-        self.identity.proto = 'gpg'
-        self.identity.host = user_id
-        self.curve_name = curve_name
-
-    def pubkey(self, ecdh=False):
-        """Return public key as VerifyingKey object."""
-        addr = util.get_bip32_address(identity=self.identity, ecdh=ecdh)
-        if ecdh:
-            curve_name = formats.get_ecdh_curve_name(self.curve_name)
-        else:
-            curve_name = self.curve_name
-        public_node = self.client_wrapper.connection.get_public_node(
-            n=addr, ecdsa_curve_name=curve_name)
-
-        return formats.decompress_pubkey(
-            pubkey=public_node.node.public_key,
-            curve_name=curve_name)
-
-    def sign(self, digest):
-        """Sign the digest and return a serialized signature."""
-        result = self.client_wrapper.connection.sign_identity(
-            identity=self.identity,
-            challenge_hidden=digest,
-            challenge_visual='',
-            ecdsa_curve_name=self.curve_name)
-        assert result.signature[:1] == b'\x00'
-        sig = result.signature[1:]
-        return (util.bytes2num(sig[:32]), util.bytes2num(sig[32:]))
-
-    def ecdh(self, pubkey):
-        """Derive shared secret using ECDH from remote public key."""
-        result = self.client_wrapper.connection.get_ecdh_session_key(
-            identity=self.identity,
-            peer_public_key=pubkey,
-            ecdsa_curve_name=formats.get_ecdh_curve_name(self.curve_name))
-        assert len(result.session_key) in {65, 33}  # NIST256 or Curve25519
-        assert result.session_key[:1] == b'\x04'
-        return result.session_key
-
-    def close(self):
-        """Close the connection to the device."""
-        self.client_wrapper.connection.close()

trezor_agent/gpg/keyring.py

@@ -24,7 +24,7 @@ def get_agent_sock_path(sp=subprocess):
 def connect_to_agent(sp=subprocess):
     """Connect to GPG agent's UNIX socket."""
     sock_path = get_agent_sock_path(sp=sp)
-    sp.check_call(['gpg-connect-agent', '/bye'])  # Stop current gpg-agent
+    sp.check_call(['gpg-connect-agent', '/bye'])  # Make sure it's running
     sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
     sock.connect(sock_path)
     return sock
@@ -158,7 +158,12 @@ def sign_digest(sock, keygrip, digest, sp=subprocess, environ=None):
     assert communicate(sock, 'SETKEYDESC '
                        'Sign+a+new+TREZOR-based+subkey') == b'OK'
     assert communicate(sock, 'PKSIGN') == b'OK'
-    line = recvline(sock).strip()
+    while True:
+        line = recvline(sock).strip()
+        if line.startswith(b'S PROGRESS'):
+            continue
+        else:
+            break
     line = unescape(line)
     log.debug('unescaped: %r', line)
     prefix, sig = line.split(b' ', 1)

trezor_agent/gpg/protocol.py

@@ -167,7 +167,9 @@ SUPPORTED_CURVES = {
 
 ECDH_ALGO_ID = 18
 
-CUSTOM_SUBPACKET = subpacket(100, b'TREZOR-GPG')  # marks "our" pubkey
+CUSTOM_KEY_LABEL = b'TREZOR-GPG'  # marks "our" pubkey
+CUSTOM_SUBPACKET_ID = 26  # use "policy URL" subpacket
+CUSTOM_SUBPACKET = subpacket(CUSTOM_SUBPACKET_ID, CUSTOM_KEY_LABEL)
 
 
 def get_curve_name_by_oid(oid):

trezor_agent/protocol.py

@@ -7,7 +7,6 @@ for more details.
 The server's source code can be found here:
 https://github.com/openssh/openssh-portable/blob/master/authfd.c
 """
-import binascii
 import io
 import logging
 
@@ -138,13 +137,13 @@ class Handler(object):
         else:
             raise KeyError('key not found')
 
-        log.debug('signing %d-byte blob', len(blob))
         label = key['name'].decode('ascii')  # label should be a string
+        log.debug('signing %d-byte blob with "%s" key', len(blob), label)
         try:
-            signature = self.signer(label=label, blob=blob)
+            signature = self.signer(blob=blob, identity=key['identity'])
         except IOError:
             return failure()
-        log.debug('signature: %s', binascii.hexlify(signature))
+        log.debug('signature: %r', signature)
 
         try:
             sig_bytes = key['verifier'](sig=signature, msg=blob)

trezor_agent/server.py

@@ -1,5 +1,6 @@
 """UNIX-domain socket server for ssh-agent implementation."""
 import contextlib
+import functools
 import logging
 import os
 import socket
@@ -42,19 +43,24 @@ def unix_domain_socket_server(sock_path):
         remove_file(sock_path)
 
 
-def handle_connection(conn, handler):
+def handle_connection(conn, handler, mutex):
     """
     Handle a single connection using the specified protocol handler in a loop.
 
+    Since this function may be called concurrently from server_thread,
+    the specified mutex is used to synchronize the device handling.
+
     Exit when EOFError is raised.
     All other exceptions are logged as warnings.
     """
     try:
         log.debug('welcome agent')
-        while True:
-            msg = util.read_frame(conn)
-            reply = handler.handle(msg=msg)
-            util.send(conn, reply)
+        with contextlib.closing(conn):
+            while True:
+                msg = util.read_frame(conn)
+                with mutex:
+                    reply = handler.handle(msg=msg)
+                util.send(conn, reply)
     except EOFError:
         log.debug('goodbye agent')
     except Exception as e:  # pylint: disable=broad-except
@@ -77,7 +83,7 @@ def retry(func, exception_type, quit_event):
             pass
 
 
-def server_thread(sock, handler, quit_event):
+def server_thread(sock, handle_conn, quit_event):
     """Run a server on the specified socket."""
     log.debug('server thread started')
 
@@ -93,8 +99,9 @@ def server_thread(sock, handler, quit_event):
         except StopIteration:
             log.debug('server stopped')
             break
-        with contextlib.closing(conn):
-            handle_connection(conn, handler)
+        # Handle connections from SSH concurrently.
+        threading.Thread(target=handle_conn,
+                         kwargs=dict(conn=conn)).start()
     log.debug('server thread stopped')
 
 
@@ -115,14 +122,23 @@ def serve(handler, sock_path=None, timeout=UNIX_SOCKET_TIMEOUT):
     If no connection is made during the specified timeout,
     retry until the context is over.
     """
+    ssh_version = subprocess.check_output(['ssh', '-V'],
+                                          stderr=subprocess.STDOUT)
+    log.debug('local SSH version: %r', ssh_version)
     if sock_path is None:
-        sock_path = tempfile.mktemp(prefix='ssh-agent-')
+        sock_path = tempfile.mktemp(prefix='trezor-ssh-agent-')
 
     environ = {'SSH_AUTH_SOCK': sock_path, 'SSH_AGENT_PID': str(os.getpid())}
+    device_mutex = threading.Lock()
     with unix_domain_socket_server(sock_path) as sock:
         sock.settimeout(timeout)
         quit_event = threading.Event()
-        kwargs = dict(sock=sock, handler=handler, quit_event=quit_event)
+        handle_conn = functools.partial(handle_connection,
+                                        handler=handler,
+                                        mutex=device_mutex)
+        kwargs = dict(sock=sock,
+                      handle_conn=handle_conn,
+                      quit_event=quit_event)
         with spawn(server_thread, kwargs):
             try:
                 yield environ

trezor_agent/tests/test_client.py

@@ -3,7 +3,7 @@ import io
 import mock
 import pytest
 
-from .. import client, factory, formats, util
+from .. import client, device, formats, util
 
 ADDR = [2147483661, 2810943954, 3938368396, 3454558782, 3848009040]
 CURVE = 'nist256p1'
@@ -12,32 +12,26 @@ PUBKEY = (b'\x03\xd8(\xb5\xa6`\xbet0\x95\xac:[;]\xdc,\xbd\xdc?\xd7\xc0\xec'
           b'\xdd\xbc+\xfar~\x9dAis')
 PUBKEY_TEXT = ('ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzd'
                'HAyNTYAAABBBNgotaZgvnQwlaw6Wztd3Cy93D/XwOzdvCv6cn6dQWlzNMEQeW'
-               'VUfhvrGljR2Z/CMRONY6ejB+9PnpUOPuzYqi8= ssh://localhost:22\n')
+               'VUfhvrGljR2Z/CMRONY6ejB+9PnpUOPuzYqi8= <localhost:22|nist256p1>\n')
 
 
-class FakeConnection(object):
+class MockDevice(device.interface.Device):  # pylint: disable=abstract-method
 
-    def __init__(self):
-        self.closed = False
+    def connect(self):  # pylint: disable=no-self-use
+        return mock.Mock()
 
     def close(self):
-        self.closed = True
+        self.conn = None
 
-    def clear_session(self):
-        self.closed = True
+    def pubkey(self, identity, ecdh=False):  # pylint: disable=unused-argument
+        assert self.conn
+        return PUBKEY
 
-    def get_public_node(self, n, ecdsa_curve_name=b'secp256k1'):
-        assert not self.closed
-        assert n == ADDR
-        assert ecdsa_curve_name in {'secp256k1', 'nist256p1'}
-        result = mock.Mock(spec=[])
-        result.node = mock.Mock(spec=[])
-        result.node.public_key = PUBKEY
-        return result
-
-    def ping(self, msg):
-        assert not self.closed
-        return msg
+    def sign(self, identity, blob):
+        """Sign given blob and return the signature (as bytes)."""
+        assert self.conn
+        assert blob == BLOB
+        return SIG
 
 
 def identity_type(**kwargs):
@@ -50,13 +44,6 @@ def identity_type(**kwargs):
     return result
 
 
-def load_client():
-    return factory.ClientWrapper(connection=FakeConnection(),
-                                 identity_type=identity_type,
-                                 device_name='DEVICE_NAME',
-                                 call_exception=Exception)
-
-
 BLOB = (b'\x00\x00\x00 \xce\xe0\xc9\xd5\xceu/\xe8\xc5\xf2\xbfR+x\xa1\xcf\xb0'
         b'\x8e;R\xd3)m\x96\x1b\xb4\xd8s\xf1\x99\x16\xaa2\x00\x00\x00\x05roman'
         b'\x00\x00\x00\x0essh-connection\x00\x00\x00\tpublickey'
@@ -66,71 +53,33 @@ BLOB = (b'\x00\x00\x00 \xce\xe0\xc9\xd5\xceu/\xe8\xc5\xf2\xbfR+x\xa1\xcf\xb0'
         b'\xdd\xbc+\xfar~\x9dAis4\xc1\x10yeT~\x1b\xeb\x1aX\xd1\xd9\x9f\xc21'
         b'\x13\x8dc\xa7\xa3\x07\xefO\x9e\x95\x0e>\xec\xd8\xaa/')
 
-SIG = (b'\x00R\x19T\xf2\x84$\xef#\x0e\xee\x04X\xc6\xc3\x99T`\xd1\xd8\xf7!'
+SIG = (b'R\x19T\xf2\x84$\xef#\x0e\xee\x04X\xc6\xc3\x99T`\xd1\xd8\xf7!'
        b'\x862@cx\xb8\xb9i@1\x1b3#\x938\x86]\x97*Y\xb2\x02Xa\xdf@\xecK'
        b'\xdc\xf0H\xab\xa8\xac\xa7? \x8f=C\x88N\xe2')
 
 
 def test_ssh_agent():
-    label = 'localhost:22'
-    c = client.Client(loader=load_client)
-    ident = c.get_identity(label=label)
-    assert ident.host == 'localhost'
-    assert ident.proto == 'ssh'
-    assert ident.port == '22'
-    assert ident.user is None
-    assert ident.path is None
-    assert ident.index == 0
+    identity = device.interface.Identity(identity_str='localhost:22',
+                                         curve_name=CURVE)
+    c = client.Client(device=MockDevice())
+    assert c.get_public_key(identity) == PUBKEY_TEXT
+    signature = c.sign_ssh_challenge(blob=BLOB, identity=identity)
 
-    with c:
-        assert c.get_public_key(label) == PUBKEY_TEXT
+    key = formats.import_public_key(PUBKEY_TEXT)
+    serialized_sig = key['verifier'](sig=signature, msg=BLOB)
 
-        def ssh_sign_identity(identity, challenge_hidden,
-                              challenge_visual, ecdsa_curve_name):
-            assert (util.identity_to_string(identity) ==
-                    util.identity_to_string(ident))
-            assert challenge_hidden == BLOB
-            assert challenge_visual == ''
-            assert ecdsa_curve_name == 'nist256p1'
+    stream = io.BytesIO(serialized_sig)
+    r = util.read_frame(stream)
+    s = util.read_frame(stream)
+    assert not stream.read()
+    assert r[:1] == b'\x00'
+    assert s[:1] == b'\x00'
+    assert r[1:] + s[1:] == SIG
 
-            result = mock.Mock(spec=[])
-            result.public_key = PUBKEY
-            result.signature = SIG
-            return result
+    # pylint: disable=unused-argument
+    def cancel_sign(identity, blob):
+        raise IOError(42, 'ERROR')
 
-        c.client.sign_identity = ssh_sign_identity
-        signature = c.sign_ssh_challenge(label=label, blob=BLOB)
-
-        key = formats.import_public_key(PUBKEY_TEXT)
-        serialized_sig = key['verifier'](sig=signature, msg=BLOB)
-
-        stream = io.BytesIO(serialized_sig)
-        r = util.read_frame(stream)
-        s = util.read_frame(stream)
-        assert not stream.read()
-        assert r[:1] == b'\x00'
-        assert s[:1] == b'\x00'
-        assert r[1:] + s[1:] == SIG[1:]
-
-        c.client.call_exception = ValueError
-
-        # pylint: disable=unused-argument
-        def cancel_sign_identity(identity, challenge_hidden,
-                                 challenge_visual, ecdsa_curve_name):
-            raise c.client.call_exception(42, 'ERROR')
-
-        c.client.sign_identity = cancel_sign_identity
-        with pytest.raises(IOError):
-            c.sign_ssh_challenge(label=label, blob=BLOB)
-
-
-def test_utils():
-    identity = mock.Mock(spec=[])
-    identity.proto = 'https'
-    identity.user = 'user'
-    identity.host = 'host'
-    identity.port = '443'
-    identity.path = '/path'
-
-    url = 'https://user@host:443/path'
-    assert util.identity_to_string(identity) == url
+    c.device.sign = cancel_sign
+    with pytest.raises(IOError):
+        c.sign_ssh_challenge(blob=BLOB, identity=identity)

trezor_agent/tests/test_factory.py

@@ -1,97 +0,0 @@
-import mock
-import pytest
-
-from .. import factory
-
-
-def test_load():
-
-    def single():
-        return [0]
-
-    def nothing():
-        return []
-
-    def double():
-        return [1, 2]
-
-    assert factory.load(loaders=[single]) == 0
-    assert factory.load(loaders=[single, nothing]) == 0
-    assert factory.load(loaders=[nothing, single]) == 0
-
-    with pytest.raises(IOError):
-        factory.load(loaders=[])
-
-    with pytest.raises(IOError):
-        factory.load(loaders=[single, single])
-
-    with pytest.raises(IOError):
-        factory.load(loaders=[double])
-
-
-def factory_load_client(**kwargs):
-    # pylint: disable=protected-access
-    return list(factory._load_client(**kwargs))
-
-
-def test_load_nothing():
-    hid_transport = mock.Mock(spec_set=['enumerate'])
-    hid_transport.enumerate.return_value = []
-    result = factory_load_client(
-        name=None,
-        client_type=None,
-        hid_transport=hid_transport,
-        passphrase_ack=None,
-        identity_type=None,
-        required_version=None,
-        call_exception=None)
-    assert result == []
-
-
-def create_client_type(version):
-    conn = mock.Mock(spec=[])
-    conn.features = mock.Mock(spec=[])
-    major, minor, patch = version.split('.')
-    conn.features.device_id = 'DEVICE_ID'
-    conn.features.label = 'LABEL'
-    conn.features.vendor = 'VENDOR'
-    conn.features.major_version = major
-    conn.features.minor_version = minor
-    conn.features.patch_version = patch
-    conn.features.revision = b'\x12\x34\x56\x78'
-    return mock.Mock(spec_set=[], return_value=conn)
-
-
-def test_load_single():
-    hid_transport = mock.Mock(spec_set=['enumerate'])
-    hid_transport.enumerate.return_value = [0]
-    for version in ('1.3.4', '1.3.5', '1.4.0', '2.0.0'):
-        passphrase_ack = mock.Mock(spec_set=[])
-        client_type = create_client_type(version)
-        client_wrapper, = factory_load_client(
-            name='DEVICE_NAME',
-            client_type=client_type,
-            hid_transport=hid_transport,
-            passphrase_ack=passphrase_ack,
-            identity_type=None,
-            required_version='>=1.3.4',
-            call_exception=None)
-        assert client_wrapper.connection is client_type.return_value
-        assert client_wrapper.device_name == 'DEVICE_NAME'
-        client_wrapper.connection.callback_PassphraseRequest('MESSAGE')
-        assert passphrase_ack.mock_calls == [mock.call(passphrase='')]
-
-
-def test_load_old():
-    hid_transport = mock.Mock(spec_set=['enumerate'])
-    hid_transport.enumerate.return_value = [0]
-    for version in ('1.3.3', '1.2.5', '1.1.0', '0.9.9'):
-        with pytest.raises(ValueError):
-            factory_load_client(
-                name='DEVICE_NAME',
-                client_type=create_client_type(version),
-                hid_transport=hid_transport,
-                passphrase_ack=None,
-                identity_type=None,
-                required_version='>=1.3.4',
-                call_exception=None)

trezor_agent/tests/test_protocol.py

@@ -1,6 +1,6 @@
 import pytest
 
-from .. import formats, protocol
+from .. import device, formats, protocol
 
 # pylint: disable=line-too-long
 
@@ -17,6 +17,7 @@ NIST256_SIGN_REPLY = b'\x00\x00\x00j\x0e\x00\x00\x00e\x00\x00\x00\x13ecdsa-sha2-
 
 def test_list():
     key = formats.import_public_key(NIST256_KEY)
+    key['identity'] = device.interface.Identity('ssh://localhost', 'nist256p1')
     h = protocol.Handler(keys=[key], signer=None)
     reply = h.handle(LIST_MSG)
     assert reply == LIST_NIST256_REPLY
@@ -28,14 +29,15 @@ def test_unsupported():
     assert reply == b'\x00\x00\x00\x01\x05'
 
 
-def ecdsa_signer(label, blob):
-    assert label == 'ssh://localhost'
+def ecdsa_signer(identity, blob):
+    assert str(identity) == '<ssh://localhost|nist256p1>'
     assert blob == NIST256_BLOB
     return NIST256_SIG
 
 
 def test_ecdsa_sign():
     key = formats.import_public_key(NIST256_KEY)
+    key['identity'] = device.interface.Identity('ssh://localhost', 'nist256p1')
     h = protocol.Handler(keys=[key], signer=ecdsa_signer)
     reply = h.handle(NIST256_SIGN_MSG)
     assert reply == NIST256_SIGN_REPLY
@@ -43,31 +45,30 @@ def test_ecdsa_sign():
 
 def test_sign_missing():
     h = protocol.Handler(keys=[], signer=ecdsa_signer)
-
     with pytest.raises(KeyError):
         h.handle(NIST256_SIGN_MSG)
 
 
 def test_sign_wrong():
-    def wrong_signature(label, blob):
-        assert label == 'ssh://localhost'
+    def wrong_signature(identity, blob):
+        assert str(identity) == '<ssh://localhost|nist256p1>'
         assert blob == NIST256_BLOB
         return b'\x00' * 64
 
     key = formats.import_public_key(NIST256_KEY)
+    key['identity'] = device.interface.Identity('ssh://localhost', 'nist256p1')
     h = protocol.Handler(keys=[key], signer=wrong_signature)
-
     with pytest.raises(ValueError):
         h.handle(NIST256_SIGN_MSG)
 
 
 def test_sign_cancel():
-    def cancel_signature(label, blob):  # pylint: disable=unused-argument
+    def cancel_signature(identity, blob):  # pylint: disable=unused-argument
         raise IOError()
 
     key = formats.import_public_key(NIST256_KEY)
+    key['identity'] = device.interface.Identity('ssh://localhost', 'nist256p1')
     h = protocol.Handler(keys=[key], signer=cancel_signature)
-
     assert h.handle(NIST256_SIGN_MSG) == protocol.failure()
 
 
@@ -79,14 +80,15 @@ ED25519_BLOB = b'''\x00\x00\x00 i3\xae}yk\\\xa1L\xb9\xe1\xbf\xbc\x8e\x87\r\x0e\x
 ED25519_SIG = b'''\x8eb)\xa6\xe9P\x83VE\xfbq\xc6\xbf\x1dV3\xe3<O\x11\xc0\xfa\xe4\xed\xb8\x81.\x81\xc8\xa6\xba\x10RA'a\xbc\xa9\xd3\xdb\x98\x07\xf0\x1a\x9c4\x84<\xaf\x99\xb7\xe5G\xeb\xf7$\xc1\r\x86f\x16\x8e\x08\x05'''  # nopep8
 
 
-def ed25519_signer(label, blob):
-    assert label == 'ssh://localhost'
+def ed25519_signer(identity, blob):
+    assert str(identity) == '<ssh://localhost|ed25519>'
     assert blob == ED25519_BLOB
     return ED25519_SIG
 
 
 def test_ed25519_sign():
     key = formats.import_public_key(ED25519_KEY)
+    key['identity'] = device.interface.Identity('ssh://localhost', 'ed25519')
     h = protocol.Handler(keys=[key], signer=ed25519_signer)
     reply = h.handle(ED25519_SIGN_MSG)
     assert reply == ED25519_SIGN_REPLY

trezor_agent/tests/test_server.py

@@ -1,3 +1,4 @@
+import functools
 import io
 import os
 import socket
@@ -37,30 +38,32 @@ class FakeSocket(object):
 
 
 def test_handle():
+    mutex = threading.Lock()
+
     handler = protocol.Handler(keys=[], signer=None)
     conn = FakeSocket()
-    server.handle_connection(conn, handler)
+    server.handle_connection(conn, handler, mutex)
 
     msg = bytearray([protocol.msg_code('SSH_AGENTC_REQUEST_RSA_IDENTITIES')])
     conn = FakeSocket(util.frame(msg))
-    server.handle_connection(conn, handler)
+    server.handle_connection(conn, handler, mutex)
     assert conn.tx.getvalue() == b'\x00\x00\x00\x05\x02\x00\x00\x00\x00'
 
     msg = bytearray([protocol.msg_code('SSH2_AGENTC_REQUEST_IDENTITIES')])
     conn = FakeSocket(util.frame(msg))
-    server.handle_connection(conn, handler)
+    server.handle_connection(conn, handler, mutex)
     assert conn.tx.getvalue() == b'\x00\x00\x00\x05\x0C\x00\x00\x00\x00'
 
     msg = bytearray([protocol.msg_code('SSH2_AGENTC_ADD_IDENTITY')])
     conn = FakeSocket(util.frame(msg))
-    server.handle_connection(conn, handler)
+    server.handle_connection(conn, handler, mutex)
     conn.tx.seek(0)
     reply = util.read_frame(conn.tx)
     assert reply == util.pack('B', protocol.msg_code('SSH_AGENT_FAILURE'))
 
     conn_mock = mock.Mock(spec=FakeSocket)
     conn_mock.recv.side_effect = [Exception, EOFError]
-    server.handle_connection(conn=conn_mock, handler=None)
+    server.handle_connection(conn=conn_mock, handler=None, mutex=mutex)
 
 
 def test_server_thread():
@@ -78,8 +81,10 @@ def test_server_thread():
         def getsockname(self):  # pylint: disable=no-self-use
             return 'fake_server'
 
+    handler = protocol.Handler(keys=[], signer=None),
+    handle_conn = functools.partial(server.handle_connection, handler=handler)
     server.server_thread(sock=FakeServer(),
-                         handler=protocol.Handler(keys=[], signer=None),
+                         handle_conn=handle_conn,
                          quit_event=quit_event)
 
 

trezor_agent/util.py

@@ -1,10 +1,8 @@
 """Various I/O and serialization utilities."""
 import binascii
 import contextlib
-import hashlib
 import io
 import logging
-import re
 import struct
 
 log = logging.getLogger(__name__)
@@ -180,59 +178,10 @@ class Reader(object):
             self._captured = None
 
 
-_identity_regexp = re.compile(''.join([
-    '^'
-    r'(?:(?P<proto>.*)://)?',
-    r'(?:(?P<user>.*)@)?',
-    r'(?P<host>.*?)',
-    r'(?::(?P<port>\w*))?',
-    r'(?P<path>/.*)?',
-    '$'
-]))
-
-
-def string_to_identity(s, identity_type):
-    """Parse string into Identity protobuf."""
-    m = _identity_regexp.match(s)
-    result = m.groupdict()
-    log.debug('parsed identity: %s', result)
-    kwargs = {k: v for k, v in result.items() if v}
-    return identity_type(**kwargs)
-
-
-def identity_to_string(identity):
-    """Dump Identity protobuf into its string representation."""
-    result = []
-    if identity.proto:
-        result.append(identity.proto + '://')
-    if identity.user:
-        result.append(identity.user + '@')
-    result.append(identity.host)
-    if identity.port:
-        result.append(':' + identity.port)
-    if identity.path:
-        result.append(identity.path)
-    return ''.join(result)
-
-
-def get_bip32_address(identity, ecdh=False):
-    """Compute BIP32 derivation address according to SLIP-0013/0017."""
-    index = struct.pack('<L', identity.index)
-    addr = index + identity_to_string(identity).encode('ascii')
-    log.debug('address string: %r', addr)
-    digest = hashlib.sha256(addr).digest()
-    s = io.BytesIO(bytearray(digest))
-
-    hardened = 0x80000000
-    addr_0 = [13, 17][bool(ecdh)]
-    address_n = [addr_0] + list(recv(s, '<LLLL'))
-    return [(hardened | value) for value in address_n]
-
-
-def setup_logging(verbosity):
+def setup_logging(verbosity, **kwargs):
     """Configure logging for this tool."""
     fmt = ('%(asctime)s %(levelname)-12s %(message)-100s '
            '[%(filename)s:%(lineno)d]')
     levels = [logging.WARNING, logging.INFO, logging.DEBUG]
     level = levels[min(verbosity, len(levels) - 1)]
-    logging.basicConfig(format=fmt, level=level)
+    logging.basicConfig(format=fmt, level=level, **kwargs)