Merge pull request #43 from ojarva/parse-authorized-files

Add support for parsing authorized_keys files

Author: ojarva

.isort.cfg

@@ -1,3 +1,3 @@
 [settings]
 line_length = 120
-
+not_skip = __init__.py

README.rst

@@ -59,6 +59,23 @@ Usage:
   print(ssh.options_raw)  # None (string of optional options at the beginning of public key)
   print(ssh.options)  # None (options as a dictionary, parsed and validated)
 
+
+Parsing of `authorized_keys` files:
+
+::
+
+  from sshpubkeys import AuthorizedKeysFile
+
+  key_file = AuthorizedKeysFile("""ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIEGODBKRjsFB/1v3pDRGpA6xR+QpOJg9vat0brlbUNDD\n"""
+             """#This is a comment\n\n"""
+             """ecdsa-sha2-nistp521 AAAAE2VjZHNhLXNoYTItbmlzdHA1MjEAAAAIbmlzdHA1MjEAAACFBAF9QpvUneTvt8"""
+             """lu0ePSuzr7iLE9ZMPu2DFTmqh7BVn89IHuQ5dfg9pArxfHZWgu9lMdlOykVx0I6OXkE35A/mFqwwApyiPmiwno"""
+             """jmRnN//pApl6QQFINHzV/PGOSi599F1Y2tHQwcdb44CPOhkUmHtC9wKazSvw/ivbxNjcMzhhHsWGnA=="""
+             strict=True, disallow_options=True)
+  for key in key_file.keys:
+      print(key.key_type, key.bits, key.hash_512())
+
+
 Options
 -------
 

sshpubkeys/__init__.py

@@ -1,2 +1,2 @@
-from .keys import *  # pylint:disable=wildcard-import
 from .exceptions import *  # pylint:disable=wildcard-import
+from .keys import *  # pylint:disable=wildcard-import

sshpubkeys/keys.py

@@ -27,13 +27,36 @@
 from cryptography.hazmat.primitives.asymmetric.dsa import DSAParameterNumbers, DSAPublicNumbers
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicNumbers
 
-from .exceptions import *  # pylint:disable=wildcard-import,unused-wildcard-import
+from .exceptions import (InvalidKeyError, InvalidKeyLengthError, InvalidOptionNameError, InvalidOptionsError,
+                         InvalidTypeError, MalformedDataError, MissingMandatoryOptionValueError, TooLongKeyError,
+                         TooShortKeyError, UnknownOptionNameError)
 
-__all__ = ["SSHKey"]
+__all__ = ["AuthorizedKeysFile", "SSHKey"]
+
+
+class AuthorizedKeysFile(object):  # pylint:disable=too-few-public-methods
+    """Represents a full authorized_keys file.
+
+    Comments and empty lines are ignored."""
+
+    def __init__(self, file_obj, **kwargs):
+        self.keys = []
+        self.parse(file_obj, **kwargs)
+
+    def parse(self, file_obj, **kwargs):
+        for line in file_obj:
+            line = line.strip()
+            if not line:
+                continue
+            if line.startswith("#"):
+                continue
+            ssh_key = SSHKey(line, **kwargs)
+            ssh_key.parse()
+            self.keys.append(ssh_key)
 
 
 class SSHKey(object):  # pylint:disable=too-many-instance-attributes
-    """Presents a single SSH keypair.
+    """Represents a single SSH keypair.
 
     ssh_key = SSHKey(key_data, strict=True)
     ssh_key.parse()
@@ -110,6 +133,9 @@ def __init__(self, keydata=None, **kwargs):
             except (InvalidKeyError, NotImplementedError):
                 pass
 
+    def __str__(self):
+        return "Key type: %s, bits: %s, options: %s" % (self.key_type, self.bits, self.options)
+
     def reset(self):
         """Reset all data fields."""
         for field in self.FIELDS:

tests/__init__.py

@@ -4,13 +4,26 @@
 
 """
 
+import sys
 import unittest
-from sshpubkeys import SSHKey
+
+from sshpubkeys import AuthorizedKeysFile, InvalidOptionsError, SSHKey
+
+from .authorized_keys import items as list_of_authorized_keys
+from .invalid_authorized_keys import items as list_of_invalid_authorized_keys
+from .invalid_keys import keys as list_of_invalid_keys
+from .invalid_options import options as list_of_invalid_options
 from .valid_keys import keys as list_of_valid_keys
 from .valid_keys_rfc4716 import keys as list_of_valid_keys_rfc4716
-from .invalid_keys import keys as list_of_invalid_keys
 from .valid_options import options as list_of_valid_options
-from .invalid_options import options as list_of_invalid_options
+
+if sys.version_info.major == 2:
+    from io import BytesIO as StringIO
+else:
+    from io import StringIO
+
+
+DEFAULT_KEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIEGODBKRjsFB/1v3pDRGpA6xR+QpOJg9vat0brlbUNDD"
 
 
 class TestMisc(unittest.TestCase):
@@ -52,6 +65,32 @@ def check_invalid_option(self, option, expected_error):
         ssh = SSHKey()
         self.assertRaises(expected_error, ssh.parse_options, option)
 
+    def test_disallow_options(self):
+        ssh = SSHKey(disallow_options=True)
+        key = """command="dump /home",no-pty,no-port-forwarding """ + DEFAULT_KEY
+        self.assertRaises(InvalidOptionsError, ssh.parse, key)
+
+
+class TestAuthorizedKeys(unittest.TestCase):
+
+    def check_valid_file(self, file_str, valid_keys_count):
+        file_obj = StringIO(file_str)
+        key_file = AuthorizedKeysFile(file_obj)
+        for item in key_file.keys:
+            self.assertIsInstance(item, SSHKey)
+        self.assertEqual(len(key_file.keys), valid_keys_count)
+
+    def check_invalid_file(self, file_str, expected_error):
+        file_obj = StringIO(file_str)
+        self.assertRaises(expected_error, AuthorizedKeysFile, file_obj)
+
+    def test_disallow_options(self):
+        file_obj = StringIO("""command="dump /home",no-pty,no-port-forwarding """ + DEFAULT_KEY)
+        self.assertRaises(InvalidOptionsError, AuthorizedKeysFile, file_obj, disallow_options=True)
+        file_obj.seek(0)
+        key_file = AuthorizedKeysFile(file_obj)
+        self.assertEqual(len(key_file.keys), 1)
+
 
 def loop_options(options):
     """ Loop over list of options and dynamically create tests """
@@ -106,11 +145,29 @@ def ch(pubkey, expected_error, **kwargs):
             setattr(TestKeys, "test_%s_mode_%s" % (prefix_tmp, mode), ch(pubkey, expected_error, **kwargs))
 
 
+def loop_authorized_keys(keyset):
+    def ch(file_str, valid_keys_count):
+        return lambda self: self.check_valid_file(file_str, valid_keys_count)
+    for i, items in enumerate(keyset):
+        prefix_tmp = "%s_%s" % (items[0], i)
+        setattr(TestAuthorizedKeys, "test_%s" % prefix_tmp, ch(items[1], items[2]))
+
+
+def loop_invalid_authorized_keys(keyset):
+    def ch(file_str, expected_error, **kwargs):
+        return lambda self: self.check_invalid_file(file_str, expected_error, **kwargs)
+    for i, items in enumerate(keyset):
+        prefix_tmp = "%s_%s" % (items[0], i)
+        setattr(TestAuthorizedKeys, "test_invalid_%s" % prefix_tmp, ch(items[1], items[2]))
+
+
 loop_valid(list_of_valid_keys, "valid_key")
 loop_valid(list_of_valid_keys_rfc4716, "valid_key_rfc4716")
 loop_invalid(list_of_invalid_keys, "invalid_key")
 loop_options(list_of_valid_options)
 loop_invalid_options(list_of_invalid_options)
+loop_authorized_keys(list_of_authorized_keys)
+loop_invalid_authorized_keys(list_of_invalid_authorized_keys)
 
 if __name__ == '__main__':
     unittest.main()

tests/authorized_keys.py

@@ -0,0 +1,8 @@
+from .valid_keys import keys
+
+items = [
+    ["empty_file", "", 0],
+    ["single_key", keys[0][0], 1],
+    ["comment_only", "# Nothing else than a comment here", 0],
+    ["lines_with_spaces", " # Comments\n  \n" + keys[0][0] + "\n#asdf", 1],
+]

tests/invalid_authorized_keys.py

@@ -0,0 +1,10 @@
+from sshpubkeys.exceptions import InvalidKeyError, MalformedDataError
+
+from .valid_keys import keys as valid_keys
+
+from.invalid_keys import keys as invalid_keys
+
+items = [
+    ["lines_with_spaces", " # Comments\n  \n" + valid_keys[0][0] + "\nasdf", InvalidKeyError],
+    ["invalid_key", "# Comments\n" + invalid_keys[0][0], MalformedDataError],
+]