util: move BIP32 address related functions

trezor_agent/__main__.py

@@ -7,14 +7,14 @@ import re
 import subprocess
 import sys
 
-from . import client, formats, protocol, server
+from . import client, formats, protocol, server, util
 
 log = logging.getLogger(__name__)
 
 
 def ssh_args(label):
     """Create SSH command for connecting specified server."""
-    identity = client.string_to_identity(label, identity_type=dict)
+    identity = util.string_to_identity(label, identity_type=dict)
 
     args = []
     if 'port' in identity:

trezor_agent/client.py

@@ -6,8 +6,6 @@ It is used for getting SSH public keys and ECDSA signing of server requests.
 import binascii
 import io
 import logging
-import re
-import struct
 
 from . import factory, formats, util
 
@@ -40,7 +38,7 @@ class Client(object):
 
     def get_identity(self, label, index=0):
         """Parse label string into Identity protobuf."""
-        identity = string_to_identity(label, self.identity_type)
+        identity = util.string_to_identity(label, self.identity_type)
         identity.proto = 'ssh'
         identity.index = index
         return identity
@@ -48,10 +46,10 @@ class Client(object):
     def get_public_key(self, label):
         """Get SSH public key corresponding to specified by label."""
         identity = self.get_identity(label=label)
-        label = identity_to_string(identity)  # canonize key 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 = get_address(identity)
+        addr = util.get_bip32_address(identity)
         node = self.client.get_public_node(n=addr,
                                            ecdsa_curve_name=self.curve)
 
@@ -93,55 +91,6 @@ class Client(object):
         return result.signature[1:]
 
 
-_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_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 = formats.hashfunc(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 _parse_ssh_blob(data):
     res = {}
     i = io.BytesIO(data)

trezor_agent/gpg/encode.py

@@ -3,7 +3,7 @@ import logging
 import time
 
 from . import decode, keyring, protocol
-from .. import client, factory, formats, util
+from .. import factory, formats, util
 
 log = logging.getLogger(__name__)
 
@@ -21,7 +21,7 @@ class HardwareSigner(object):
 
     def pubkey(self, ecdh=False):
         """Return public key as VerifyingKey object."""
-        addr = client.get_address(identity=self.identity, ecdh=ecdh)
+        addr = util.get_bip32_address(identity=self.identity, ecdh=ecdh)
         public_node = self.client_wrapper.connection.get_public_node(
             n=addr, ecdsa_curve_name=self.curve_name)
 

trezor_agent/tests/test_client.py

@@ -87,8 +87,8 @@ def test_ssh_agent():
 
         def ssh_sign_identity(identity, challenge_hidden,
                               challenge_visual, ecdsa_curve_name):
-            assert (client.identity_to_string(identity) ==
-                    client.identity_to_string(ident))
+            assert (util.identity_to_string(identity) ==
+                    util.identity_to_string(ident))
             assert challenge_hidden == BLOB
             assert challenge_visual == ''
             assert ecdsa_curve_name == 'nist256p1'
@@ -133,4 +133,4 @@ def test_utils():
     identity.path = '/path'
 
     url = 'https://user@host:443/path'
-    assert client.identity_to_string(identity) == url
+    assert util.identity_to_string(identity) == url

trezor_agent/util.py

@@ -1,9 +1,14 @@
 """Various I/O and serialization utilities."""
 import binascii
 import contextlib
+import hashlib
 import io
+import logging
+import re
 import struct
 
+log = logging.getLogger(__name__)
+
 
 def send(conn, data):
     """Send data blob to connection socket."""
@@ -173,3 +178,52 @@ class Reader(object):
             yield
         finally:
             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]