Support SSH signatures

https://www.agwa.name/blog/post/ssh_signatures

See here for more details:
https://github.com/openssh/openssh-portable/blob/master/PROTOCOL.sshsig
https://github.com/openssh/openssh-portable/blob/master/sshsig.c
2a9c9f7272

libagent/ssh/client.py

@@ -31,33 +31,50 @@ class Client:
 
     def sign_ssh_challenge(self, blob, identity):
         """Sign given blob using a private key on the device."""
-        msg = _parse_ssh_blob(blob)
+        log.debug('blob: %r', blob)
-        log.debug('%s: user %r via %r (%r)',
+        msg = parse_ssh_blob(blob)
-                  msg['conn'], msg['user'], msg['auth'], msg['key_type'])
+        if msg['sshsig']:
-        log.debug('nonce: %r', msg['nonce'])
+            log.info('please confirm "%s" signature for "%s" using %s...',
-        fp = msg['public_key']['fingerprint']
+                     msg['namespace'], identity.to_string(), self.device)
-        log.debug('fingerprint: %s', fp)
+        else:
-        log.debug('hidden challenge size: %d bytes', len(blob))
+            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...',
+            log.info('please confirm user "%s" login to "%s" using %s...',
-                 msg['user'].decode('ascii'), identity.to_string(),
+                     msg['user'].decode('ascii'), identity.to_string(),
-                 self.device)
+                     self.device)
 
         with self.device:
             return self.device.sign(blob=blob, identity=identity)
 
 
-def _parse_ssh_blob(data):
+def parse_ssh_blob(data):
+    """Parse binary data into a dict."""
     res = {}
-    i = io.BytesIO(data)
+    if data.startswith(b'SSHSIG'):
-    res['nonce'] = util.read_frame(i)
+        i = io.BytesIO(data[6:])
-    i.read(1)  # SSH2_MSG_USERAUTH_REQUEST == 50 (from ssh2.h, line 108)
+        # https://github.com/openssh/openssh-portable/blob/master/PROTOCOL.sshsig
-    res['user'] = util.read_frame(i)
+        res['sshsig'] = True
-    res['conn'] = util.read_frame(i)
+        res['namespace'] = util.read_frame(i)
-    res['auth'] = util.read_frame(i)
+        res['reserved'] = util.read_frame(i)
-    i.read(1)  # have_sig == 1 (from sshconnect2.c, line 1056)
+        res['hashalg'] = util.read_frame(i)
-    res['key_type'] = util.read_frame(i)
+        res['message'] = util.read_frame(i)
-    public_key = util.read_frame(i)
+    else:
-    res['public_key'] = formats.parse_pubkey(public_key)
+        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)
+
     assert not i.read()
     return res

libagent/ssh/protocol.py

@@ -144,6 +144,10 @@ class Handler:
             signature = self.conn.sign(blob=blob, identity=key['identity'])
         except IOError:
             return failure()
+        except Exception:
+            log.exception('signature with "%s" key failed', label)
+            raise
+
         log.debug('signature: %r', signature)
 
         try:

libagent/ssh/tests/test_client.py

@@ -74,3 +74,52 @@ def test_ssh_agent():
     c.device.sign = cancel_sign
     with pytest.raises(IOError):
         c.sign_ssh_challenge(blob=BLOB, identity=identity)
+
+
+CHALLENGE_BLOB = (
+    b'\x00\x00\x00 \xe4\x08\x8e"J#\x83 \x05\x90\x1e\xa9\xf9C\xb1\xd2\x8f\xc3\x8c\xea\xd8\xf6E'
+    b'%q\xff\x07\xfa\xd8\x8b\xdf\xbd2\x00\x00\x00\x03git\x00\x00\x00\x0essh-connection\x00\x00'
+    b'\x00\tpublickey\x01\x00\x00\x00\x0bssh-ed25519\x00\x00\x003\x00\x00\x00\x0bssh-ed25519'
+    b'\x00\x00\x00 \xd1q\x1ab\xc6\xf0d\x19\xe2q<\x05\x0b\xdao\xa1\xcb\xae\xad\xc9\x0b\x16\xf3'
+    b'\xc2m\x84q8qU\xda\xb0'
+)
+
+
+def test_parse_ssh_challenge():
+    result = client.parse_ssh_blob(CHALLENGE_BLOB)
+    result['public_key'].pop('verifier')
+    assert result == {
+        'auth': b'publickey',
+        'conn': b'ssh-connection',
+        'key_type': b'ssh-ed25519',
+        'nonce': b'\xe4\x08\x8e"J#\x83 \x05\x90\x1e\xa9\xf9C\xb1\xd2\x8f\xc3\x8c\xea'
+                 b'\xd8\xf6E%q\xff\x07\xfa\xd8\x8b\xdf\xbd',
+        'public_key': {'blob': b'\x00\x00\x00\x0bssh-ed25519\x00\x00\x00 \xd1'
+                               b'q\x1ab\xc6\xf0d\x19\xe2q<\x05\x0b\xdao\xa1\xcb'
+                               b'\xae\xad\xc9\x0b\x16\xf3\xc2m\x84q8qU\xda\xb0',
+                       'curve': 'ed25519',
+                       'fingerprint': '47:a3:26:af:0b:5d:a2:c3:91:ed:26:36:94:be:3a:d5',
+                       'type': b'ssh-ed25519'},
+        'sshsig': False,
+        'user': b'git',
+    }
+
+
+FILE_SIG_BLOB = (
+    b"SSHSIG\x00\x00\x00\x04file\x00\x00\x00\x00\x00\x00\x00\x06sha512\x00\x00\x00@r\xb7r\xfeM"
+    b"\xe5w\xf0#w\x1dbl\xca\to=\x90\xb69\xd1:u{\xe5\xe4\xf1\xb1\xa8C\xb8\xfcM\x91\x9f\x12\xa8"
+    b"\x1d`\x00\x848C<\x85\x8e\xf0o\xdab\xdcQ\xce\xf2\xda\xc3\xae\xa9\x1e%\x85\xcd\xe3'"
+)
+
+
+def test_parse_ssh_signature():
+    result = client.parse_ssh_blob(FILE_SIG_BLOB)
+    assert result == {
+        'hashalg': b'sha512',
+        'message': b'r\xb7r\xfeM\xe5w\xf0#w\x1dbl\xca\to=\x90\xb69\xd1:u{'
+                   b'\xe5\xe4\xf1\xb1\xa8C\xb8\xfcM\x91\x9f\x12\xa8\x1d`\x00\x848C<'
+                   b"\x85\x8e\xf0o\xdab\xdcQ\xce\xf2\xda\xc3\xae\xa9\x1e%\x85\xcd\xe3'",
+        'namespace': b'file',
+        'reserved': b'',
+        'sshsig': True,
+    }