aswi2021jmsd-gitlab4/proj/services/cryptography.py @ 57898b2f

import subprocess

# encryption method to be used when generating private keys
from proj.utils.temporary_file import TemporaryFile

PRIVATE_KEY_ENCRYPTION_METHOD = "-aes256"

# openssl executable name
OPENSSL_EXECUTABLE = "openssl"


class CryptographyService:

    @staticmethod
    def __subject_to_param_format(subject):
        """
        Converts the given subject to a dictionary containing openssl field names mapped to subject's fields
        :param subject: subject to be converted
        :return: a dictionary containing openssl field names mapped to subject's fields
        """
        subj_dict = {}
        if subject.common_name is not None:
            subj_dict["CN"] = subject.common_name
        if subject.country is not None:
            subj_dict["C"] = subject.country
        if subject.locality is not None:
            subj_dict["L"] = subject.locality
        if subject.state is not None:
            subj_dict["ST"] = subject.state
        if subject.organization is not None:
            subj_dict["O"] = subject.organization
        if subject.organization_unit is not None:
            subj_dict["OU"] = subject.organization_unit
        if subject.email_address is not None:
            subj_dict["emailAddress"] = subject.email_address

        # merge the subject into a "subj" parameter format
        return "".join([f"/{key}={value}" for key, value in subj_dict.items()])

    @staticmethod
    def _run_for_output(args=None, proc_input=None, executable=OPENSSL_EXECUTABLE):
        """
        Launches a new process in which the given executable is run. STDIN and process arguments can be set.
        If the process ends with a non-zero then <CryptographyException> is raised.

        :param args: Arguments to be passed to the program.
        :param proc_input: String input to be passed to the stdin of the created process.
        :param executable: Executable to be run (defaults to openssl)
        :return: If the process ends with a zero return code then the STDOUT of the process is returned as a byte array.
        """
        if args is None:
            args = []
        try:
            # prepend the name of the executable
            args.insert(0, executable)

            # create a new process
            proc = subprocess.Popen(args, stdin=subprocess.PIPE if proc_input is not None else None,
                                    stdout=subprocess.PIPE,
                                    stderr=subprocess.PIPE)

            out, err = proc.communicate(proc_input)

            if proc.returncode != 0:
                # if the process did not result in zero result code, then raise an exception
                if err is not None and len(err) > 0:
                    raise CryptographyException(executable, args, err.decode())
                else:
                    raise CryptographyException(executable, args,
                                                f""""Execution resulted in non-zero argument""")

            return out
        except FileNotFoundError:
            raise CryptographyException(executable, args, f""""{executable}" not found in the current PATH.""")

    def create_private_key(self, passphrase=None):
        """
        Creates a private key with the option to encrypt it using a passphrase.
        :param passphrase: A passphrase to be used when encrypting the key (if none is passed then the key is not
        encrypted at all). Empty passphrase ("") also results in a key that is not encrypted.
        :return: string containing the generated private key in PEM format
        """
        if passphrase is None or len(passphrase) == 0:
            return self._run_for_output(["genrsa", "2048"]).decode()
        else:
            return self._run_for_output(
                ["genrsa", PRIVATE_KEY_ENCRYPTION_METHOD, "-passout", f"pass:{passphrase}", "2048"]).decode()

    def create_sscrt(self, key, subject, config="", extensions="", key_pass=None):
        """
        Creates a root CA

        :param key: private key of the CA to be used
        :param subject: an instance of <Subject> representing the subject to be added to the certificate
        :param config: string containing the configuration to be used
        :param extensions: name of the section in the configuration representing extensions
        :param key_pass: passphrase of the private key

        :return: string containing the generated certificate in PEM format
        """
        assert key is not None
        assert subject is not None

        subj = self.__subject_to_param_format(subject)

        with TemporaryFile("openssl.conf", config) as conf_path:
            args = ["req", "-x509", "-new", "-subj", subj,
                    "-key", "-"]
            if len(config) > 0:
                args.extend(["-config", conf_path])

            if len(extensions) > 0:
                args.extend(["-extensions", extensions])

            # it would be best to not send the pass phrase at all, but for some reason pytest then prompts for
            # the pass phrase (this does not happen when run from pycharm)

            #  add the passphrase even when None is passed. Otherwise when running tests with pytest some tests freeze
            # waiting for the passphrase to be typed in
            args.extend(["-passin", f"pass:{key_pass}"])

            return self._run_for_output(args, proc_input=bytes(key, encoding="utf-8")).decode()

    def __create_csr(self, subject, subject_key, subject_key_pass=""):
        """
        Creates a CSR (Certificate Signing Request)

        :param subject: an instance of <Subject> representing the subject to be added to the CSR
        :param subject_key: the private key of the subject to be used to generate the CSR
        :param subject_key_pass: passphrase of the subject's private key
        :return: string containing the generated certificate signing request in PEM format
        """

        subj_param = self.__subject_to_param_format(subject)

        args = ["req", "-new", "-subj", subj_param, "-key", "-"]

        # add the passphrase even when None is passed. Otherwise when running tests with pytest some tests freeze
        # waiting for the passphrase to be typed in
        args.extend(["-passin", f"pass:{subject_key_pass}"])

        return self._run_for_output(args, proc_input=bytes(subject_key, encoding="utf-8")).decode()

    def __sign_csr(self, csr, issuer_pem, issuer_key, issuer_key_pass=None, extensions="", days=30):
        """
        Signs the given CSR by the given issuer CA

        :param csr: a string containing the CSR to be signed
        :param issuer_pem: string containing the certificate of the issuer that will sign this CSR in PEM format
        :param issuer_key: string containing the private key of the issuer's certificate in PEM format
        :param issuer_key_pass: string containing the passphrase of the private key of the issuer's certificate in PEM
        format
        :param extensions: extensions to be applied when signing the CSR
        :param days: number of days for which the certificate will be valid
        :return: string containing the generated and signed certificate in PEM format
        """

        # concatenate CSR, issuer certificate and issuer's key (will be used in the openssl call)
        proc_input = csr + issuer_pem + issuer_key

        # prepare openssl parameters...
        # CSR, CA and CA's private key will be passed via stdin (that's the meaning of the '-' symbol)
        params = ["x509", "-req", "-in", "-", "-CA", "-", "-CAkey", "-", "-CAcreateserial", "-days", str(days)]

        # TODO delete created -.srl file

        with TemporaryFile("extensions.conf", extensions) as ext_path:
            # add the passphrase even when None is passed. Otherwise when running tests with pytest some tests freeze
            # waiting for the passphrase to be typed in
            params.extend(["-passin", f"pass:{issuer_key_pass}"])

            if len(extensions) > 0:
                params.extend(["-extfile", ext_path])

            return self._run_for_output(params, proc_input=(bytes(proc_input, encoding="utf-8"))).decode()

    def create_crt(self, subject, subject_key, issuer_pem, issuer_key, subject_key_pass=None, issuer_key_pass=None,
                   extensions="",
                   days=30):
        """
        Creates a certificate by using the given subject, subject's key, issuer and its key.

        :param subject: subject to be added to the created certificate
        :param subject_key: string containing the private key to be used when creating the certificate in PEM format
        :param issuer_key: string containing the private key of the issuer's certificate in PEM format
        :param issuer_pem: string containing the certificate of the issuer that will sign this CSR in PEM format
        :param issuer_key: string containing the private key of the issuer's certificate in PEM format
        :param subject_key_pass: string containing the passphrase of the private key used when creating the certificate in PEM
        format
        :param issuer_key_pass: string containing the passphrase of the private key of the issuer's certificate in PEM
        format
        :param extensions: extensions to be applied when creating the certificate
        :param days: number of days for which the certificate will be valid
        :return: string containing the generated certificate in PEM format
        """
        csr = self.__create_csr(subject, subject_key, subject_key_pass=subject_key_pass)
        return self.__sign_csr(csr, issuer_pem, issuer_key, issuer_key_pass=issuer_key_pass, extensions=extensions,
                               days=days)

    @staticmethod
    def verify_cert(certificate):
        """
        Verifies whether the given certificate is not expired.

        :param certificate: certificate to be verified in PEM format
        :return: Returns `true` if the certificate is not expired, `false` when expired.
        """
        # call openssl to check whether the certificate is valid to this date
        args = [OPENSSL_EXECUTABLE, "x509", "-checkend", "0", "-noout", "-text", "-in", "-"]

        # create a new process
        proc = subprocess.Popen(args, stdin=subprocess.PIPE,
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE)

        out, err = proc.communicate(bytes(certificate, encoding="utf-8"))

        # zero return code means that the certificate is valid
        if proc.returncode == 0:
            return True
        elif proc.returncode == 1 and "Certificate will expire" in out.decode():
            # 1 return code means that the certificate is invalid but such message has to be present in the proc output
            return False
        else:
            # the process failed because of some other reason (incorrect cert format)
            raise CryptographyException(OPENSSL_EXECUTABLE, args, err.decode())


class CryptographyException(Exception):

    def __init__(self, executable, args, message):
        self.executable = executable
        self.args = args
        self.message = message

    def __str__(self):
        return f"""
        EXECUTABLE: {self.executable}
        ARGS: {self.args}
        MESSAGE: {self.message}
        """