[pull] master from romanz:master

libagent/formats.py

@@ -225,7 +225,7 @@ def export_public_key(vk, label):
 def import_public_key(line):
     """Parse public key textual format, as saved at a .pub file."""
     log.debug('loading SSH public key: %r', line)
-    file_type, base64blob, name = line.split()
+    file_type, base64blob, name = line.strip().split(maxsplit=2)
     blob = base64.b64decode(base64blob)
     result = parse_pubkey(blob)
     result['name'] = name.encode('utf-8')

libagent/ssh/client.py

@@ -10,6 +10,11 @@
 
 log = logging.getLogger(__name__)
 
+SUPPORTED_CERT_TYPES = {
+    formats.SSH_ED25519_CERT_TYPE,
+    formats.SSH_NIST256_CERT_TYPE,
+}
+
 
 class Client:
     """Client wrapper for SSH authentication device."""
@@ -31,22 +36,17 @@ def export_public_keys(self, identities):
 
     def sign_ssh_challenge(self, blob, identity):
         """Sign given blob using a private key on the device."""
-        log.debug('blob: %r', blob)
+        log.debug('blob (%d bytes): %r', len(blob), blob)
         msg = parse_ssh_blob(blob)
+        log.debug('parsed: %r', msg)
+
+        identity_str = identity.to_string()
         if msg['sshsig']:
             log.info('please confirm "%s" signature for "%s" using %s...',
-                     msg['namespace'], identity.to_string(), self.device)
+                     msg['namespace'], identity_str, self.device)
         else:
-            log.debug('%s: user %r via %r (%r)',
-                      msg['conn'], msg['user'], msg['auth'], msg['key_type'])
-            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'), identity.to_string(),
-                     self.device)
+            log.info('please confirm "%s" signature for "%s" using %s...',
+                     msg['key_type'].decode('ascii'), identity_str, self.device)
 
         with self.device:
             return self.device.sign(blob=blob, identity=identity)
@@ -66,17 +66,45 @@ def parse_ssh_blob(data):
     else:
         i = io.BytesIO(data)
         res['sshsig'] = False
-        res['nonce'] = util.read_frame(i)
-        i.read(1)  # SSH2_MSG_USERAUTH_REQUEST == 50 (from ssh2.h, line 108)
-        res['user'] = util.read_frame(i)
-        res['conn'] = util.read_frame(i)
-        res['auth'] = util.read_frame(i)
-        i.read(1)  # have_sig == 1 (from sshconnect2.c, line 1056)
-        res['key_type'] = util.read_frame(i)
-        public_key = util.read_frame(i)
-        res['public_key'] = formats.parse_pubkey(public_key)
+        first_frame = util.read_frame(i)
+        # https://github.com/openssh/openssh-portable/blob/master/PROTOCOL.certkeys
+        is_cert = first_frame in SUPPORTED_CERT_TYPES
+        if is_cert:
+            # see `sshkey_certify_custom()` for details:
+            # https://github.com/openssh/openssh-portable/blob/master/sshkey.c
+            res['key_type'] = first_frame
+            res['nonce'] = util.read_frame(i)
+            if first_frame == formats.SSH_NIST256_CERT_TYPE:
+                res['curve'] = util.read_frame(i)
+            res['pubkey'] = util.read_frame(i)
+            res['serial_number'] = util.recv(i, '>Q')
+            res['type'] = util.recv(i, '>L')
+            res['key_id'] = util.read_frame(i)
+            res['valid_principals'] = tuple(_iter_parse_list(util.read_frame(i)))
+            res['valid_after'] = util.recv(i, '>Q')
+            res['valid_before'] = util.recv(i, '>Q')
+            res['critical_options'] = tuple(_iter_parse_list(util.read_frame(i)))
+            res['extensions'] = tuple(_iter_parse_list(util.read_frame(i)))
+            res['reserved'] = util.read_frame(i)
+            res['signature_key'] = util.read_frame(i)
+        else:
+            res['nonce'] = first_frame
+            i.read(1)  # SSH2_MSG_USERAUTH_REQUEST == 50 (from ssh2.h, line 108)
+            res['user'] = util.read_frame(i)
+            res['conn'] = util.read_frame(i)
+            res['auth'] = util.read_frame(i)
+            i.read(1)  # have_sig == 1 (from sshconnect2.c, line 1056)
+            res['key_type'] = util.read_frame(i)
+            public_key = util.read_frame(i)
+            res['public_key'] = formats.parse_pubkey(public_key)
 
     unparsed = i.read()
     if unparsed:
         log.warning('unparsed blob: %r', unparsed)
     return res
+
+
+def _iter_parse_list(blob):
+    i = io.BytesIO(blob)
+    while i.tell() < len(blob):
+        yield util.read_frame(i)

libagent/tests/test_formats.py

@@ -60,7 +60,7 @@ def test_fingerprint():
     'E2VjZHNhLXNoYTItbmlzdHAyNTYAAABIAAAAICUMX1taTy6'
     'y+1Aa1m7kXHI/Qv7ZZIeNp7ndmCRLFCSuAAAAIBaX43k0Ye'
     'Bk8a5zp6FyFCBYVOtis/DUbGm07d7miPnE '
-    'hello\n'
+    'hello world\n'
 )
 
 _public_key_ed25519_cert_BLOB = (
@@ -139,7 +139,7 @@ def test_parse_ed25519():
 
 def test_parse_ed25519_cert():
     p = formats.import_public_key(_public_key_ed25519_cert)
-    assert p['name'] == b'hello'
+    assert p['name'] == b'hello world'
     assert p['curve'] == 'ed25519'
 
     assert p['blob'] == _public_key_ed25519_cert_BLOB