Improve SSL mangement: (#127)

- Support passing private key and certificate files as strings
- Change edge/ssl module to use crypto services
- Update edge/ssl UT

Author: saimonation

cterasdk/core/ssl.py

@@ -2,7 +2,7 @@
 from zipfile import ZipFile
 
 from .base_command import BaseCommand
-from ..lib import FileSystem, X509Certificate, TempfileServices, create_certificate_chain
+from ..lib import FileSystem, X509Certificate, PrivateKey, TempfileServices, create_certificate_chain
 
 
 class SSL(BaseCommand):
@@ -50,23 +50,25 @@ def create_zip_archive(self, private_key, *certificates):
         """
         Create a ZIP archive that can be imported to CTERA Portal
 
-        :param str private_key: A path to the PEM-encoded private key file
-        :param list[str] certificates: A list of paths of the PEM-encoded certificate files
+        :param str private_key: The PEM-encoded private key, or a path to the PEM-encoded private key file
+        :param list[str] certificates: The PEM-encoded certificates, or a list of paths of the PEM-encoded certificate files
         """
         tempdir = TempfileServices.mkdir()
 
         key_basename = 'private.key'
-        private_keyfile = self._filesystem.copyfile(private_key, FileSystem.join(tempdir, key_basename))
+        key_object = PrivateKey.load_private_key(private_key)
+        key_filepath = FileSystem.join(tempdir, key_basename)
+        self._filesystem.write(key_filepath, key_object.pem_data)
 
         cert_basename = 'certificate'
-        certificates = [X509Certificate.from_file(certificate) for certificate in certificates]
+        certificates = [X509Certificate.load_certificate(certificate) for certificate in certificates]
         certificate_chain = create_certificate_chain(*certificates)
 
         certificate_chain_zip_archive = None
         if certificate_chain:
             certificate_chain_zip_archive = FileSystem.join(tempdir, '{}.zip'.format(cert_basename))
             with ZipFile(certificate_chain_zip_archive, 'w') as zip_archive:
-                zip_archive.write(private_keyfile, key_basename)
+                zip_archive.write(key_filepath, key_basename)
                 for idx, certificate in enumerate(certificate_chain):
                     filename = '{}{}.crt'.format(cert_basename, idx if idx > 0 else '')
                     filepath = FileSystem.join(tempdir, filename)
@@ -87,8 +89,8 @@ def import_from_chain(self, private_key, *certificates):
         """
         Import an SSL Certificate to CTERA Portal from a chain
 
-        :param str private_key: A path to the PEM-encoded private key file
-        :param list[str] certificates: A list of paths to the PEM-encoded certificates
+        :param str private_key: The PEM-encoded private key, or a path to the PEM-encoded private key file
+        :param list[str] certificates: The PEM-encoded certificates, or a list of paths of the PEM-encoded certificate files
         """
         zipflie = self.create_zip_archive(private_key, *certificates)
         return self.import_from_zip(zipflie)

cterasdk/edge/ssl.py

@@ -1,14 +1,12 @@
 import logging
 
 from .base_command import BaseCommand
-from ..lib import FileSystem
+from ..lib import X509Certificate, PrivateKey, create_certificate_chain
 from ..common import Object
 
 
 class SSL(BaseCommand):
-    """ Gateway SSL APIs """
-
-    BEGIN_PEM = '-----BEGIN'
+    """ Edge Filer SSL APIs """
 
     def enable_http(self):
         """
@@ -52,48 +50,34 @@ def get_storage_ca(self):
         """
         return self._gateway.get('/status/extStorageTrustedCA')
 
-    def set_storage_ca(self, certificate):
+    def import_storage_ca(self, certificate):
         """
-        Set object storage trusted CA certificate
+        Import the object storage trusted CA certificate
 
         :param str certificate: The PEM-encoded certificate or a path to the PEM-encoded server certificate file
         """
         logging.getLogger().info('Setting trusted object storage CA certificate')
         param = Object()
         param._classname = 'ExtTrustedCA'  # pylint: disable=protected-access
-        param.certificate = SSL._obtain_secret(certificate)
-        return self._gateway.put('/config/extStorageTrustedCA', param)
+        param.certificate = X509Certificate.load_certificate(certificate).pem_data.decode('utf-8')
+        logging.getLogger().info("Uploading object storage certificate.")
+        response = self._gateway.put('/config/extStorageTrustedCA', param)
+        logging.getLogger().info("Uploaded object storage certificate.")
+        return response
 
-    def set_certificate(self, private_key, *certificates):
+    def import_certificate(self, private_key, *certificates):
         """
-        Set the Edge Filer's web server's certificate.
+        Import the Edge Filer's web server's SSL certificate
 
-        :param str private_key: The PEM-encoded private key or a path to the PEM-encoded private key file
-        :param list[str] certificates: The PEM-encoded certificates or a path to the PEM-encoded certificates
+        :param str private_key: The PEM-encoded private key, or a path to the PEM-encoded private key file
+        :param list[str] certificates: The PEM-encoded certificates, or a list of paths to the PEM-encoded certificates
         """
-        logging.getLogger().debug('Loading private key')
-        certificate_chain = [SSL._obtain_secret(private_key)]
-        logging.getLogger().debug('Loading certificates')
-        certificate_chain = certificate_chain + [SSL._obtain_secret(certificate) for certificate in certificates]
 
-        logging.getLogger().info("Uploading certificate chain")
-        server_certificate = '\n' + '\n'.join(certificate_chain).replace('\n\n', '\n')
-        response = self._gateway.put('/config/certificate', server_certificate)
-        logging.getLogger().info("Uploaded certificate chain")
+        key_object = PrivateKey.load_private_key(private_key)
+        certificates = [X509Certificate.load_certificate(certificate) for certificate in certificates]
+        certificate_chain = [certificate.pem_data.decode('utf-8') for certificate in create_certificate_chain(*certificates)]
+        server_certificate = ''.join([key_object.pem_data.decode('utf-8')] + certificate_chain)
+        logging.getLogger().info("Uploading SSL certificate.")
+        response = self._gateway.put('/config/certificate', "\n{}".format(server_certificate))
+        logging.getLogger().info("Uploaded SSL certificate.")
         return response
-
-    @staticmethod
-    def _obtain_secret(secret):
-        if not secret.startswith(SSL.BEGIN_PEM):
-            file_info, secret = SSL._file_contents(secret)
-            logging.getLogger().debug(
-                "Reading file. %s", {'name': file_info['name'], 'size': file_info['size'], 'type': file_info['mimetype']}
-            )
-        return secret
-
-    @staticmethod
-    def _file_contents(filepath):
-        file_info = FileSystem.instance().get_local_file_info(filepath)
-        with open(filepath, 'r') as f:
-            file_content = f.read()
-        return (file_info, file_content)

cterasdk/lib/__init__.py

@@ -6,4 +6,4 @@
 from .file_access_base import FileAccessBase  # noqa: E402, F401
 from .filesystem import FileSystem  # noqa: E402, F401
 from .tracker import track, ErrorStatus  # noqa: E402, F401
-from .crypto import CryptoServices, X509Certificate, create_certificate_chain  # noqa: E402, F401
+from .crypto import CryptoServices, X509Certificate, PrivateKey, create_certificate_chain  # noqa: E402, F401

cterasdk/lib/crypto.py

@@ -6,9 +6,10 @@
 from cryptography import x509
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import rsa
-from cryptography.hazmat.primitives.serialization import Encoding, PrivateFormat, PublicFormat, NoEncryption
+from cryptography.hazmat.primitives.serialization import Encoding, PrivateFormat, PublicFormat, NoEncryption, load_pem_private_key
 
 from .. import config
+from ..exception import LocalFileNotFound
 from .filesystem import FileSystem
 
 
@@ -68,6 +69,41 @@ def compare_certificates(a, b):
     return 0
 
 
+class PrivateKey:
+
+    def __init__(self, private_key):
+        self.private_key = private_key
+
+    @property
+    def pem_data(self):
+        return self.private_key.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption())
+
+    @staticmethod
+    def from_file(path, password=None):
+        with open(path, 'r') as f:
+            data = f.read()
+        return PrivateKey.from_string(data, password)
+
+    @staticmethod
+    def from_string(data, password=None):
+        return PrivateKey.from_bytes(data.encode('utf-8'), password)
+
+    @staticmethod
+    def from_bytes(data, password=None):
+        return PrivateKey(load_pem_private_key(data, password))
+
+    @staticmethod
+    def load_private_key(key, password=None):
+        if isinstance(key, bytes):
+            return PrivateKey.from_bytes(key, password)
+
+        try:
+            FileSystem.instance().get_local_file_info(key)
+            return PrivateKey.from_file(key, password)
+        except LocalFileNotFound:
+            return PrivateKey.from_string(key, password)
+
+
 class X509Certificate:
 
     def __init__(self, certificate):
@@ -112,6 +148,17 @@ def from_string(data):
     def from_bytes(data):
         return X509Certificate(x509.load_pem_x509_certificate(data))
 
+    @staticmethod
+    def load_certificate(cert):
+        if isinstance(cert, bytes):
+            return X509Certificate.from_bytes(cert)
+
+        try:
+            FileSystem.instance().get_local_file_info(cert)
+            return X509Certificate.from_file(cert)
+        except LocalFileNotFound:
+            return X509Certificate.from_string(cert)
+
     def __str__(self):
         return str(
             dict(

tests/ut/test_edge_ssl.py

@@ -9,11 +9,13 @@ class TestEdgeSSL(base_edge.BaseEdgeTest):
 
     def setUp(self):
         super().setUp()
-        self._private_key = './certs/private.key'
+        self._key_filepath = './certs/private.key'
         self._domain_cert = './certs/certificate.crt'
-        self._intermediate = './certs/intermediate.crt'
-        self._ca = './certs/ca.crt'
-        self._certificate = ssl.SSL.BEGIN_PEM
+        self._intermediate_cert = './certs/intermediate.crt'
+        self._root_cert = './certs/ca.crt'
+
+        self._private_key_contents = 'private_key'
+        self._certificate_contents = 'certificate'
 
     def test_is_http_disabled_true(self):
         get_response = True
@@ -53,14 +55,30 @@ def test_disable_http(self):
         ssl.SSL(self._filer).disable_http()
         self._filer.put.assert_called_once_with('/config/fileservices/webdav/forceHttps', True)
 
-    def test_set_certificate(self):
-        data = 'data\n'
-        self._init_filer()
-        self.patch_call("cterasdk.edge.ssl.FileSystem.get_local_file_info")
-        mock_open = mock.mock_open(read_data=data)
-        with mock.patch("builtins.open", mock_open):
-            ssl.SSL(self._filer).set_certificate(self._private_key, self._domain_cert, self._intermediate, self._ca)
-            self._filer.put.assert_called_once_with('/config/certificate', '\n' + data * 4)
+    def test_import_certificate(self):
+        put_response = 'Success'
+        self._init_filer(put_response=put_response)
+        mock_load_private_key = self.patch_call("cterasdk.lib.crypto.PrivateKey.load_private_key")
+        mock_load_private_key.return_value = TestEdgeSSL._get_secret(self._private_key_contents)
+        mock_load_certificate = self.patch_call("cterasdk.lib.crypto.X509Certificate.load_certificate")
+        mock_load_certificate.return_value = TestEdgeSSL._get_secret(self._certificate_contents)
+        ret = ssl.SSL(self._filer).import_certificate(self._key_filepath, self._domain_cert, self._intermediate_cert, self._root_cert)
+        mock_load_private_key.assert_called_once_with(self._key_filepath)
+        mock_load_certificate.assert_has_calls(
+            [
+                mock.call(self._domain_cert),
+                mock.call(self._intermediate_cert),
+                mock.call(self._root_cert),
+            ]
+        )
+        expected_param = ''.join([
+            self._private_key_contents,
+            self._certificate_contents,
+            self._certificate_contents,
+            self._certificate_contents
+        ])
+        self._filer.put.assert_called_once_with('/config/certificate', '\n{}'.format(expected_param))
+        self.assertEqual(ret, put_response)
 
     def test_get_storage_ca(self):
         get_response = 'Success'
@@ -69,13 +87,16 @@ def test_get_storage_ca(self):
         self._filer.get.assert_called_once_with('/status/extStorageTrustedCA')
         self.assertEqual(ret, get_response)
 
-    def test_set_storage_ca(self):
+    def test_import_storage_ca(self):
         put_response = 'Success'
         self._init_filer(put_response=put_response)
-        ret = ssl.SSL(self._filer).set_storage_ca(self._certificate)
+        mock_load_certificate = self.patch_call("cterasdk.lib.crypto.X509Certificate.load_certificate")
+        mock_load_certificate.return_value = TestEdgeSSL._get_secret(self._certificate_contents)
+        ret = ssl.SSL(self._filer).import_storage_ca(self._domain_cert)
         expected_param = Object()
         expected_param._classname = 'ExtTrustedCA'  # pylint: disable=protected-access
-        expected_param.certificate = self._certificate
+        expected_param.certificate = self._certificate_contents
+        mock_load_certificate.assert_called_once_with(self._domain_cert)
         self._filer.put.assert_called_once_with('/config/extStorageTrustedCA', mock.ANY)
         actual_param = self._filer.put.call_args[0][1]
         self._assert_equal_objects(actual_param, expected_param)
@@ -86,3 +107,11 @@ def test_remove_storage_ca(self):
         self._init_filer(put_response=put_response)
         ssl.SSL(self._filer).remove_storage_ca()
         self._filer.put.assert_called_once_with('/config/extStorageTrustedCA', None)
+
+    @staticmethod
+    def _get_secret(secret):
+        param = Object()
+        param.pem_data = secret.encode('utf-8')
+        param.subject = 'subject'
+        param.issuer = 'issuer'
+        return param