Merge pull request #11 from ojarva/keyhandling-updates

Keyhandling updates

README.rst

@@ -7,7 +7,7 @@ OpenSSH Public Key Parser for Python
 .. image:: https://pypip.in/v/sshpubkeys/badge.png
     :target: https://pypi.python.org/pypi/sshpubkeys
 
-This library validates OpenSSH public keys.
+Native implementation for validating OpenSSH public keys.
 
 Currently ssh-rsa, ssh-dss (DSA), ssh-ed25519 and ecdsa keys with NIST curves are supported.
 
@@ -27,16 +27,32 @@ Usage:
 
 ::
 
+  import sys
   from sshpubkeys import SSHKey
+
   ssh = SSHKey("ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAYQCxO38tKAJXIs9ivPxt7AY"
         "dfybgtAR1ow3Qkb9GPQ6wkFHQqcFDe6faKCxH6iDRteo4D8L8B"
         "xwzN42uZSB0nfmjkIxFTcEU3mFSXEbWByg78aoddMrAAjatyrh"
-        "H1pON6P0= ojarva@ojar-laptop")
+        "H1pON6P0= ojarva@ojar-laptop", strict_mode=True)
+  try:
+      ssh.parse()
+  except InvalidKeyException as err:
+      print("Invalid key:", err)
+      sys.exit(1)
+  except NotImplementedError as err:
+      print("Invalid key type:", err)
+      sys.exit(1)
+
   print(ssh.bits)  # 768
   print(ssh.hash_md5())  # 56:84:1e:90:08:3b:60:c7:29:70:5f:5e:25:a6:3b:86
   print(ssh.hash_sha256())  # SHA256:xk3IEJIdIoR9MmSRXTP98rjDdZocmXJje/28ohMQEwM
   print(ssh.hash_sha512())  # SHA512:1C3lNBhjpDVQe39hnyy+xvlZYU3IPwzqK1rVneGavy6O3/ebjEQSFvmeWoyMTplIanmUK1hmr9nA8Skmj516HA
 
+Options
+-------
+
+- strict_mode: if set to True, disallows keys OpenSSH's ssh-keygen refuses to create. For instance, this includes DSA keys where length != 1024 bits and RSA keys shorter than 1024-bit. If set to False, tries to allow all keys OpenSSH accepts, including highly insecure 1-bit DSA keys.
+
 Exceptions
 ----------
 

setup.py

@@ -9,7 +9,7 @@
 
 setup(
     name='sshpubkeys',
-    version='1.2.2',
+    version='2.0.0',
     description='SSH public key parser',
     long_description=long_description,
     url='https://github.com/ojarva/python-sshpubkeys',

sshpubkeys/__init__.py

@@ -63,15 +63,20 @@ class MalformedDataException(InvalidKeyException):
 
 class SSHKey(object):  # pylint:disable=too-many-instance-attributes
     """
-    ssh_key = SSHKey(key_data)
+    ssh_key = SSHKey(key_data, strict=True)
     ssh_key.parse()
+
+    strict=True (default) only allows keys ssh-keygen generates. Setting strict mode to false allows
+    all keys OpenSSH actually accepts, including highly insecure ones. For example, OpenSSH accepts
+    512-bit DSA keys and 64-bit RSA keys which are highly insecure.
     """
 
-    VALID_DSA_PARAMETERS = [
-        (1024, 160),
-        (2048, 160),
-        (3072, 160),
-    ]
+    DSA_MIN_LENGTH_STRICT = 1024
+    DSA_MAX_LENGTH_STRICT = 1024
+    DSA_MIN_LENGTH_LOOSE = 1
+    DSA_MAX_LENGTH_LOOSE = 16384
+
+    DSA_N_LENGTH = 160
 
     ECDSA_CURVE_DATA = {
         b"nistp256": (ecdsa.curves.NIST256p, hashlib.sha256),
@@ -81,11 +86,14 @@ class SSHKey(object):  # pylint:disable=too-many-instance-attributes
         b"nistp521": (ecdsa.curves.NIST521p, hashlib.sha512),
     }
 
-    RSA_MIN_LENGTH = 768
-    RSA_MAX_LENGTH = 16384
+    RSA_MIN_LENGTH_STRICT = 1024
+    RSA_MAX_LENGTH_STRICT = 16384
+    RSA_MIN_LENGTH_LOOSE = 768
+    RSA_MAX_LENGTH_LOOSE = 16384
+
     INT_LEN = 4
 
-    def __init__(self, keydata):
+    def __init__(self, keydata, **kwargs):
         self.keydata = keydata
         self.current_position = 0
         self.decoded_key = None
@@ -94,15 +102,19 @@ def __init__(self, keydata):
         self.ecdsa = None
         self.bits = None
         self.key_type = None
-        self.parse()
+        self.strict_mode = bool(kwargs.get("strict", True))
+        try:
+            self.parse()
+        except (InvalidKeyException, NotImplementedError):
+            pass
 
     def hash(self):
         """ Calculate md5 fingerprint.
 
         Deprecated, use .hash_md5() instead.
         """
         warnings.warn("hash() is deprecated. Use hash_md5() or hash_sha256() instead.")
-        return self.hash_md5()
+        return self.hash_md5().replace(b"MD5:", b"")
 
     def hash_md5(self):
         """ Calculate md5 fingerprint.
@@ -112,7 +124,7 @@ def hash_md5(self):
         For specification, see RFC4716, section 4.
         """
         fp_plain = hashlib.md5(self.decoded_key).hexdigest()
-        return ':'.join(a + b for a, b in zip(fp_plain[::2], fp_plain[1::2]))
+        return "MD5:" + ':'.join(a + b for a, b in zip(fp_plain[::2], fp_plain[1::2]))
 
     def hash_sha256(self):
         """ Calculate sha256 fingerprint. """
@@ -171,6 +183,8 @@ def _split_key(cls, data):
                     # Data begins after the first space
                     data = data[i + 1:]
                     break
+            else:
+                raise MalformedDataException("Couldn't find beginning of the key data")
         key_parts = data.strip().split(None, 3)
         if len(key_parts) < 2:  # Key type and content are mandatory fields.
             raise InvalidKeyException("Unexpected key format: at least type and base64 encoded value is required")
@@ -200,10 +214,16 @@ def _process_ssh_rsa(self):
         self.rsa = RSA.construct((unpacked_n, unpacked_e))
         self.bits = self.rsa.size() + 1
 
-        if self.bits < self.RSA_MIN_LENGTH:
+        if self.strict_mode:
+            min_length = self.RSA_MIN_LENGTH_STRICT
+            max_length = self.RSA_MAX_LENGTH_STRICT
+        else:
+            min_length = self.RSA_MIN_LENGTH_LOOSE
+            max_length = self.RSA_MAX_LENGTH_LOOSE
+        if self.bits < min_length:
             raise TooShortKeyException("%s key data can not be shorter than %s bits (was %s)" % (self.key_type, min_length, self.bits))
-        if self.bits > self.RSA_MAX_LENGTH:
-            raise TooLongKeyException("%s key data can not be longer than %s bits (was %s)" % (self.key_type, min_length, self.bits))
+        if self.bits > max_length:
+            raise TooLongKeyException("%s key data can not be longer than %s bits (was %s)" % (self.key_type, max_length, self.bits))
 
     def _process_ssh_dss(self):
         """ Parses ssh-dsa public keys """
@@ -215,10 +235,18 @@ def _process_ssh_dss(self):
         self.bits = self.dsa.size() + 1
 
         q_bits = self._bits_in_number(data_fields["q"])
-        for p, q in self.VALID_DSA_PARAMETERS:
-            if self.bits == p and q_bits == q:
-                return
-        raise InvalidKeyException("Incorrect DSA key parameters: bits(p)=%s, q=%s" % (self.bits, q_bits))
+        if q_bits != self.DSA_N_LENGTH:
+            raise InvalidKeyException("Incorrect DSA key parameters: bits(p)=%s, q=%s" % (self.bits, q_bits))
+        if self.strict_mode:
+            min_length = self.DSA_MIN_LENGTH_STRICT
+            max_length = self.DSA_MAX_LENGTH_STRICT
+        else:
+            min_length = self.DSA_MIN_LENGTH_LOOSE
+            max_length = self.DSA_MAX_LENGTH_LOOSE
+        if self.bits < min_length:
+            raise TooShortKeyException("%s key can not be shorter than %s bits (was %s)" % (self.key_type, min_length, self.bits))
+        if self.bits > max_length:
+            raise TooLongKeyException("%s key data can not be longer than %s bits (was %s)" % (self.key_type, max_length, self.bits))
 
     def _process_ecdsa_sha(self):
         """ Parses ecdsa-sha public keys """

tests/__init__.py

@@ -12,41 +12,52 @@
 
 
 class TestKeys(unittest.TestCase):
-    def check_key(self, pubkey, bits, fingerprint_md5, fingerprint_sha256):
+    def check_key(self, pubkey, bits, fingerprint_md5, fingerprint_sha256, **kwargs):
         """ Checks valid key """
-        ssh = SSHKey(pubkey)
+        ssh = SSHKey(pubkey, **kwargs)
+        ssh.parse()
         self.assertEqual(ssh.bits, bits)
         self.assertEqual(ssh.hash_md5(), fingerprint_md5)
         if fingerprint_sha256 is not None:
             self.assertEqual(ssh.hash_sha256(), fingerprint_sha256)
 
-    def check_fail(self, pubkey, expected_error):
+    def check_fail(self, pubkey, expected_error, **kwargs):
         """ Checks that key check raises specified exception """
         # Don't use with statement here - it does not work with Python 2.6 unittest module
-        self.assertRaises(expected_error, SSHKey, pubkey)
+        ssh_key = SSHKey(pubkey, **kwargs)
+        self.assertRaises(expected_error, ssh_key.parse)
 
 
 def loop_valid(keyset, prefix):
     """ Loop over list of valid keys and dynamically create tests """
+    def ch(pubkey, bits, fingerprint_md5, fingerprint_sha256, **kwargs):
+        return lambda self: self.check_key(pubkey, bits, fingerprint_md5, fingerprint_sha256, **kwargs)
     for i, items in enumerate(keyset):
-        def ch(pubkey, bits, fingerprint_md5, fingerprint_sha256):
-            return lambda self: self.check_key(pubkey, bits, fingerprint_md5, fingerprint_sha256)
-        prefix_tmp = "%s_%s" % (prefix, i)
-        prefix_tmp = items.pop()
-        pubkey, bits, fingerprint_md5, fingerprint_sha256 = items
-        setattr(TestKeys, "test_%s" % prefix_tmp, ch(pubkey, bits, fingerprint_md5, fingerprint_sha256))
+        modes = items.pop()
+        prefix_tmp = "%s_%s" % (prefix, items.pop())
+        for mode in modes:
+            if mode == "strict":
+                kwargs = {"strict": True}
+            else:
+                kwargs = {"strict": False}
+            pubkey, bits, fingerprint_md5, fingerprint_sha256 = items
+            setattr(TestKeys, "test_%s_mode_%s" % (prefix_tmp, mode), ch(pubkey, bits, fingerprint_md5, fingerprint_sha256, **kwargs))
 
 
 def loop_invalid(keyset, prefix):
     """ Loop over list of invalid keys and dynamically create tests """
+    def ch(pubkey, expected_error, **kwargs):
+        return lambda self: self.check_fail(pubkey, expected_error, **kwargs)
     for i, items in enumerate(keyset):
-        def ch(pubkey, expected_error):
-            return lambda self: self.check_fail(pubkey, expected_error)
-        prefix_tmp = "%s_%s" % (prefix, i)
-        if len(items) == 3:  # If there is an extra item, use that as test name.
-            prefix_tmp = items.pop()
-        pubkey, expected_error = items
-        setattr(TestKeys, "test_%s" % prefix_tmp, ch(pubkey, expected_error))
+        modes = items.pop()
+        prefix_tmp = "%s_%s" % (prefix, items.pop())
+        for mode in modes:
+            if mode == "strict":
+                kwargs = {"strict": True}
+            else:
+                kwargs = {"strict": False}
+            pubkey, expected_error = items
+            setattr(TestKeys, "test_%s_mode_%s" % (prefix_tmp, mode), ch(pubkey, expected_error, **kwargs))
 
 loop_valid(list_of_valid_keys, "valid_key")
 loop_valid(list_of_valid_keys_rfc4716, "valid_key_rfc4716")