Da

✏️ Editing: ssl_utils.py

from parallels.core import messages
from socket import gethostname
import os

from parallels.core.registry import Registry
from parallels.core.runners.windows.local import LocalWindowsRunner
from parallels.core.utils.common import mkdir_p, open_no_inherit
from parallels.core.utils.common.string_utils import safe_utf8_decode

class SSLKeys(object):
    """Global single set of SSL keys for Windows RPC agents"""

_instance = None

@classmethod
    def get_instance(cls):
        """
        Obtain instance of registry (singleton) and return it
        :rtype: parallels.core.utils.ssl_utils.SSLKeys
        """
        if cls._instance is None:
            cls._instance = SSLKeys()
        return cls._instance

def generate_keys_lazy(self):
        files = [
            self.migrator_server_crt_filename,
            self.migrator_server_key_filename,
            self.source_server_crt_filename,
            self.source_server_key_filename,
            self._keys_generated_filename
        ]
        if not all([os.path.exists(filename) for filename in files]):
            self._generate_keys()

def _generate_keys(self):
        self._generate_key_pair(
            cert_file=self.migrator_server_crt_filename,
            key_file=self.migrator_server_key_filename
        )
        self._generate_key_pair(
            cert_file=self.source_server_crt_filename,
            key_file=self.source_server_key_filename
        )
        runner = LocalWindowsRunner()
        runner.upload_file_content(
            self._keys_generated_filename,
            messages.IS_FILE_INDICATING_THAT_SSL_CERTIFICATES
        )

@property
    def migrator_server_crt_filename(self):
        return self._key_path('migration-node.crt')

@property
    def migrator_server_key_filename(self):
        return self._key_path('migration-node.key')

@property
    def source_server_crt_filename(self):
        return self._key_path('source-node.crt')

@property
    def source_server_key_filename(self):
        return self._key_path('source-node.key')

@property
    def _keys_generated_filename(self):
        return self._key_path('ssl-keys-generated')

@staticmethod
    def _key_path(base_name):
        directory = os.path.join(Registry.get_instance().get_var_dir(), 'rpc-agent', 'certificates')
        if not os.path.exists(directory):
            mkdir_p(directory)

return os.path.join(directory, base_name)

def _generate_key_pair(self, cert_file, key_file):
        key = self._create_private_key()
        cert = self._create_certificate(key)

self._save_private_key(key, key_file)
        self._save_certificate(cert, cert_file)

@staticmethod
    def _create_private_key():
        from OpenSSL import crypto
        key = crypto.PKey()
        key.generate_key(crypto.TYPE_RSA, 2048)
        return key

@staticmethod
    def _create_certificate(key):
        from OpenSSL import crypto
        cert = crypto.X509()
        cert.get_subject().C = "US"
        cert.get_subject().ST = "Virginia"
        cert.get_subject().L = "Herndon"
        cert.get_subject().O = "Parallels, Inc."
        cert.get_subject().OU = "Parallels Panel"
        cert.get_subject().CN = gethostname()
        cert.set_serial_number(1)
        cert.set_notBefore(
            # Certificate validity period begins date in the past to avoid issues with date on server,
            # in a format appropriate RFC-5280 (https://tools.ietf.org/html/rfc5280#section-4.1.2.5)
            b'20000101000000Z'
        )
        year = 365 * 24 * 60 * 60
        cert.gmtime_adj_notAfter(
            # Valid for 10 years - should be enough for any migration and date fixes on server
            10 * year
        )
        cert.set_issuer(cert.get_subject())
        cert.set_pubkey(key)
        cert.sign(key, 'sha1')
        return cert

@staticmethod
    def _save_private_key(key, key_file):
        from OpenSSL import crypto
        with open_no_inherit(key_file, "wt") as fp:
            crypto_dump_privatekey = safe_utf8_decode(crypto.dump_privatekey(crypto.FILETYPE_PEM, key))
            fp.write(crypto_dump_privatekey)

@staticmethod
    def _save_certificate(cert, cert_file):
        from OpenSSL import crypto
        with open_no_inherit(cert_file, "wt") as fp:
            crypto_dump_certificate = safe_utf8_decode(crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
            fp.write(crypto_dump_certificate)

def is_ssl_validation_error(e):
    return 'CERTIFICATE_VERIFY_FAILED' in repr(e).upper()

📁 Sid Gifari File Manager