aswi2021jmsd-gitlab/src/services/cryptography.py @ a2965e9a

import re
import subprocess
import time
import random

from src.constants import CRL_CONFIG
from src.model.certificate import Certificate
from src.model.private_key import PrivateKey
from src.model.subject import Subject
from src.utils.logger import Logger
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
        """

        Logger.debug("Function launched.")

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

        Logger.debug("Function launched.")

        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:
                    Logger.error("CryptographyException")
                    raise CryptographyException(executable, args, err.decode())
                else:
                    Logger.error("CryptographyException")
                    raise CryptographyException(executable, args,
                                                f""""Execution resulted in non-zero argument""")

            return out
        except FileNotFoundError:
            Logger.error("CryptographyException")
            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
        """

        Logger.debug("Function launched.")

        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 self signed certificate

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

        Logger.debug("Function launched.")

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

        Logger.debug("Function launched.")

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

        Logger.debug("Function launched.")

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

        Logger.debug("Function launched.")

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

        Logger.debug("Function launched.")

        # 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)
            Logger.error("CryptographyException")
            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
        """

        Logger.debug("Function launched.")

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

        Logger.debug("Function launched.")

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

        Logger.debug("Function launched.")

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

        Logger.debug("Function launched.")

        return self.__run_for_output(["version"]).decode("utf-8")

    def generate_crl(self, cert: Certificate, key: PrivateKey, index_file_path: str) -> str:
        """
        Generate a CertificateRevocationList for a specified
        certificate authority.

        :param key: key that is used to sign the CRL (must belong to the given certificate)
        :param cert: Certificate of the certificate authority that issue the CRL
        :param index_file_path: path to a file that contains the openssl index with all revoked certificates
        :return: CRL encoded in PEM format string
        """

        Logger.debug("Function launched.")

        # openssl ca requires the .srl file to exists, therefore a dummy, unused file is created
        with TemporaryFile("serial.srl", "0") as serial_file, \
             TemporaryFile("crl.conf", CRL_CONFIG % (index_file_path, serial_file)) as config_file, \
             TemporaryFile("certificate.pem", cert.pem_data) as cert_file, \
             TemporaryFile("private_key.pem", key.private_key) as key_file:

            args = ["ca", "-config", config_file, "-gencrl", "-keyfile", key_file, "-cert", cert_file, "-outdir", "."]

            if key.password is not None and key.password != "":
                args.extend(["-passin", f"pass:{key.password}"])

            return self.__run_for_output(args).decode("utf-8")

    def generate_ocsp(self, cert, key, index_path, der_ocsp_request):
        """
        Generate an OCSP Response from an OCSP Request given the issuer cert, issuer cert key and the index file.
        The OSCP Response is signed by the CA itself (recommended way according to multiple sources).

        :param cert: issuer certificate
        :param key: corresponding key
        :param index_path: path/to/the/generated/index/file
        :param der_ocsp_request: DER encoded OCSP Request
        :return: DER encoded OCSP Response
        """

        Logger.debug("Function launched.")

        with TemporaryFile("certificate.pem", cert.pem_data) as ca_certificate, \
             TemporaryFile("private_key.pem", key.private_key) as key_file, \
             TemporaryFile("request.der", der_ocsp_request) as request_file:

            args = ["ocsp", "-index", index_path, "-CA", ca_certificate, "-rsigner", ca_certificate, "-rkey", key_file,
                    "-reqin", request_file, "-respout", "-"]

            if key.password is not None and key.password != "":
                args.extend(["-passin", f"pass:{key.password}"])

            return self.__run_for_output(args)


class CryptographyException(Exception):

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

    def __str__(self):
        # TODO check log is valid here
        msg = f"""
        EXECUTABLE: {self.executable}
        ARGS: {self.args}
        MESSAGE: {self.message}
        """

        Logger.error(msg)
        return msg