aswi2021jmsd-gitlab4/src/services/cryptography.py @ d55d7f0d

import re
import subprocess
import time
import random

from src.model.subject import Subject
from src.utils.temporary_file import TemporaryFile

# encryption method to be used when generating private keys
PRIVATE_KEY_ENCRYPTION_METHOD = "-aes256"

# openssl executable name
OPENSSL_EXECUTABLE = "openssl"

# format of NOT_BEFORE NOT_AFTER date fields
NOT_AFTER_BEFORE_DATE_FORMAT = "%b %d %H:%M:%S %Y %Z"

# minimal configuration file to be used for openssl req command
# specifies distinguished_name that references empty section only
# openssl requires this option to be present
MINIMAL_CONFIG_FILE = "[req]\ndistinguished_name = req_distinguished_name\n[req_distinguished_name]\n\n"

# section to be used to specify extensions when creating a SSCRT
SSCRT_SECTION = "sscrt_ext"

CA_EXTENSIONS = "basicConstraints=critical,CA:TRUE"

# upper bound of the range of random serial numbers to be generated
MAX_SN = 4294967296


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, subject, key, config="", extensions="", key_pass=None, days=30, sn: int = None):
        """
        Creates a root CA

        :param subject: an instance of <Subject> representing the subject to be added to the certificate
        :param key: private key of the CA to be used
        :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
        :param days: number of days for which the certificate will be valid
        :param sn: serial number to be set, when "None" is set a random serial number is generated

        :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)

        # To specify extension for creating a SSCRT, one has to use a configuration
        # file instead of an extension file. Therefore the following code creates
        # the most basic configuration file with sscrt_ext section, that is later
        # reference in openssl req command using -extensions option.
        extensions += "\n"+CA_EXTENSIONS
        if len(config) == 0:
            config += MINIMAL_CONFIG_FILE
        config += "\n[ " + SSCRT_SECTION + " ]" + "\n" + extensions

        with TemporaryFile("openssl.conf", config) as conf_path:
            args = ["req", "-x509", "-new", "-subj", subj, "-days", f"{days}",
                    "-key", "-"]

            # serial number passed, use it when generating the certificate,
            # without passing it openssl generates a random one
            if sn is not None:
                args.extend(["-set_serial", str(sn)])

            if len(config) > 0:
                args.extend(["-config", conf_path])
            if len(extensions) > 0:
                args.extend(["-extensions", SSCRT_SECTION]) # when creating SSCRT, section references section in config

            # 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, key, key_pass=""):
        """
        Creates a CSR (Certificate Signing Request)

        :param subject: an instance of <Subject> representing the subject to be added to the CSR
        :param key: the private key of the subject to be used to generate the CSR
        :param 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:{key_pass}"])

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

    def __sign_csr(self, csr, issuer_pem, issuer_key, issuer_key_pass=None, extensions="", days=30, sn: int = None):
        """
        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
        :param sn: serial number to be set, when "None" is set a random serial number is generated
        :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

        # TODO find a better way to generate a random serial number or let openssl generate a .srl file
        # when serial number is not passed generate a random one
        if sn is None:
            sn = random.randint(0, MAX_SN)

        # 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),
                  "-set_serial", str(sn)]

        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,
                   sn: int = None):
        """
        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 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
        :param sn: serial number to be set, when "None" is set a random serial number is generated
        :return: string containing the generated certificate in PEM format
        """
        csr = self.__create_csr(subject, subject_key, key_pass=subject_key_pass)
        return self.__sign_csr(csr, issuer_pem, issuer_key, issuer_key_pass=issuer_key_pass, extensions=extensions,
                               days=days, sn=sn)

    @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())

    def extract_public_key_from_private_key(self, private_key_pem: str, passphrase=None) -> str:
        """
        Extracts a public key from the given private key passed in PEM format
        :param private_key_pem: PEM data representing the private key from which a public key should be extracted
        :param passphrase: passphrase to be provided when the supplied private key is encrypted
        :return: a string containing the extracted public key in PEM format
        """
        args = ["rsa", "-in", "-", "-pubout"]
        if passphrase is not None:
            args.extend(["-passin", f"pass:{passphrase}"])
        return self.__run_for_output(args, proc_input=bytes(private_key_pem, encoding="utf-8")).decode()

    def extract_public_key_from_certificate(self, cert_pem: str) -> str:
        """
        Extracts a public key from the given certificate passed in PEM format
        :param cert_pem: PEM data representing a certificate from which a public key should be extracted
        :return: a string containing the extracted public key in PEM format
        """
        # extracting public key from a certificate does not seem to require a passphrase even when
        # signed using an encrypted PK
        args = ["x509", "-in", "-", "-noout", "-pubkey"]
        return self.__run_for_output(args, proc_input=bytes(cert_pem, encoding="utf-8")).decode()

    def parse_cert_pem(self, cert_pem):
        """
        Parses the given certificate in PEM format and returns the subject of the certificate and it's NOT_BEFORE
        and NOT_AFTER field
        :param cert_pem: a certificated in a PEM format to be parsed
        :return: a tuple containing a subject, NOT_BEFORE and NOT_AFTER dates
        """
        # run openssl x509 to view certificate content
        args = ["x509", "-noout", "-subject", "-startdate", "-enddate", "-in", "-"]

        cert_info_raw = self.__run_for_output(args, proc_input=bytes(cert_pem, encoding="utf-8")).decode()

        # split lines
        results = re.split("\n", cert_info_raw)
        subj_line = results[0].strip()
        not_before_line = results[1].strip()
        not_after_line = results[2].strip()

        # attempt to extract subject via regex
        match = re.search(r"subject=(.*)", subj_line)
        if match is None:
            # TODO use logger
            print(f"Could not find subject to parse: {subj_line}")
            return None
        else:
            # find all attributes (key = value)
            found = re.findall(r"\s?([^c=\s]+)\s?=\s?([^,\n]+)", match.group(1))
            subj = Subject()
            for key, value in found:
                if key == "C":
                    subj.country = value.strip()
                elif key == "ST":
                    subj.state = value.strip()
                elif key == "L":
                    subj.locality = value.strip()
                elif key == "O":
                    subj.organization = value.strip()
                elif key == "OU":
                    subj.organization_unit = value.strip()
                elif key == "CN":
                    subj.common_name = value.strip()
                elif key == "emailAddress":
                    subj.email_address = value.strip()

        # extract notBefore and notAfter date fields
        not_before = re.search(r"notBefore=(.*)", not_before_line)
        not_after = re.search(r"notAfter=(.*)", not_after_line)

        # if date fields are found parse them into date objects
        if not_before is not None:
            not_before = time.strptime(not_before.group(1).strip(), NOT_AFTER_BEFORE_DATE_FORMAT)
        if not_after is not None:
            not_after = time.strptime(not_after.group(1).strip(), NOT_AFTER_BEFORE_DATE_FORMAT)

        # TODO wrapper class?
        # return it as a tuple
        return subj, not_before, not_after

    def get_openssl_version(self) -> str:
        """
        Get version of the OpenSSL installed on the system
        :return: version of the OpenSSL as returned from the process
        """
        return self.__run_for_output(["version"]).decode("utf-8")


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}
        """