aswi2021jmsd-gitlab/proj/services/cryptography.py @ c0aed2f5

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 _run_for_output(args=None, stdin=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 stdin: 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 stdin is not None else None, stdout=subprocess.PIPE,
                                    stderr=subprocess.PIPE)

            out, err = proc.communicate(stdin)

            if proc.returncode != 0:
                # if the process did not result in zero result code, then raise an exception
                if err is not None:
                    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: A text representation of the generated private key.
        """
        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_passphrase=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_passphrase: passphrase of the private key

        :return: byte array containing the generated certificate
        """
        assert key is not None
        assert subject is not None

        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
        subj = "".join([f"/{key}={value}" for key, value in subj_dict.items()])

        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)

            # if key_passphrase is not None:
            args.extend(["-passin", f"pass:{key_passphrase}"])

            return self._run_for_output(args, stdin=bytes(key, encoding="utf-8")).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}
        """