aswi2021jmsd-gitlab4/openssl_poc.py @ d28c02f7

import subprocess
import os
import sys

# directory where all certificates/keys will be stored
WORK_DIR = "static/openssl"

# name of the config file to be generated used when accepting a CSR
ACCEPT_CSR_CONF_FILE = "accept_csr.cnf"

# line to be inserted into the configuration file to be used when accepting a CA CSR
CA_CNF_LINE = "basicConstraints=critical,CA:TRUE"


# makes a private key without a passphrase (potentially insecure)
def make_private_key():
    return subprocess.check_output(["openssl", "genrsa", "2048"], stderr=subprocess.STDOUT)


def make_root_ca(name, key, days):
    """
    Creates a root CA

    :param name: name of the root CA to be used (will be passed to the Common Name field)
    :param key: private key of the CA to be used
    :param days: number of days for which the CA will be considered as valid
    :return: byte array containing the generated certificate
    """
    return subprocess.check_output(
        ["openssl", "req", "-new", "-x509", "-days", str(days), "-subj", f"/CN={name}", "-key", "-"],
        input=bytes(key, encoding="utf-8"), stderr=subprocess.STDOUT
    )


def make_csr(key, name):
    """
    Makes a CSR (Certificate Signing Request)
    :param key: the private key to be used to generate the certificate
    :param name: name of the certificate (will be passed to the Common Name field)
    :return: byte array containing the generated certificate signing request
    """
    return subprocess.check_output(
        ["openssl", "req", "-new", "-subj", f"/CN={name}", "-key", "-"],
        input=bytes(key, encoding="utf-8"), stderr=subprocess.STDOUT
    )


def sign_csr(csr, ca_cert, ca_key, days, config=""):
    """
    Signs the given CSR by the given CA
    :param csr: A string containing the CSR to be accepted
    :param ca_cert: A string containing the certificate of the issuer that will sign this CSR
    :param ca_key: A string containing the private key of the issuer's certificate
    :param days: The number of days for which the generated certificate will be considered as valid
    :param config: A string containing the configuration of extensions to be used
    :return: Byte array containing the generated and signed certificate
    """

    # check whether any config has been passed
    config_used = len(config) > 0
    config_file = get_path(ACCEPT_CSR_CONF_FILE)
    if config_used:
        # if config is to be applied the store it temporarily in a special file
        # that will be later passed to the openssl program
        puts(config, config_file)

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

    if config_used:
        # if config is to be applied add the generated configuration file path to the openssl program via parameters
        params.append("-extfile")
        params.append(config_file)

    # TODO delete any created files
    return subprocess.check_output(params, input=(bytes(csr + ca_cert + ca_key, encoding="utf-8")),
                                   stderr=subprocess.STDOUT)


def get_aia_cnf_line(uri):
    """
     Generates a line to be used in a configuration of X509 extensions that specifies that the current certificate will
     use Authority Information Access field to generate the chain of trust.
    :param uri: URI where the parent certificate lies
    :return:  A line to be used in a configuration of X509 extensions
    """
    return f"authorityInfoAccess = caIssuers;URI:{uri}"


# returns a path of a file in a work directory
def get_path(file):
    return f"{WORK_DIR}/{file}"


# copies the given string to the file specified by a path
def puts(content, file):
    with open(file, "w") as file:
        file.write(content)


# reads all content of a file specified by a path
def read_file(file):
    with open(file, "r") as file:
        return file.read()


def setup_root_intermediate_cas():
    """
    Generates a root and an intermediate certificate authorities
    """

    # generate a private key to be used to generate a root CA
    root_key = make_private_key().decode()
    # use the generated key to create a root CA
    root_ca = make_root_ca("ROOT CA", root_key, 1825).decode()

    # generate a private key to be used to generate an intermediate CA
    intermediate_key = make_private_key().decode()
    # use the generated key to make a CSR for a intermediate CA
    intermediate_csr = make_csr(intermediate_key, "INTER CA").decode()
    # sign the generated CSR by root CA
    # specify AIA field and that this certificate is a CA
    intermediate_ca = sign_csr(intermediate_csr, root_ca, root_key, 1096,
                               f"""
                    {get_aia_cnf_line("http://localhost:5000/static/openssl/root.crt")}
                    {CA_CNF_LINE}
                    """).decode()

    # store the generated root CA certificate/private key to files
    puts(root_ca, get_path("root.crt"))
    puts(root_key, get_path("root.key"))

    # store the generated intermediate CA certificate/private key to files
    puts(intermediate_ca, get_path("inter.crt"))
    puts(intermediate_key, get_path("inter.key"))

    # return the generated intermediate CA certificate
    return intermediate_ca, intermediate_key


# USAGE (create a chain of trust that consists of root and inter. CAs and a generated certificate signed by int. CA):
#
# "python3 openssl_poc.py MyTestCertificate"
#
# the generated certificate will have the Common Name field set to "MyTestCertificate"
# and will be stored to MyTestCertificate.crt
if __name__ == "__main__":
    # a name of the child certificate to be generated must be passed to the program via arguments
    if len(sys.argv) > 1:
        cert_name = sys.argv[1]

        # ensure that WORK_DIR directory exists
        try:
            os.mkdir(WORK_DIR)
        except IOError:
            pass

        # the program will generate a child certificate that is signed by an intermediate CA

        # root CA certificate, intermediate CA certificate and the generated user specified child certificate will be
        # save to static/openssl directory

        # if any of the files required to create such certificate is missing then the program will regenerate the whole
        # chain of trust (root/inter. CA cert/key)

        try:
            # try to read all required files (exception is thrown if any of the files are non-existent)
            read_file(get_path("root.crt"))
            inter_ca = read_file(get_path("inter.crt"))
            inter_key = read_file(get_path("inter.key"))
        except (IOError, FileNotFoundError):
            print("Creating ROOT and INTERMEDIATE CAs...")
            # any of the required files is missing - generate new ones
            (inter_ca, inter_key) = setup_root_intermediate_cas()

        # make a private key to be used for generating a child certificate
        child_key = make_private_key().decode()

        #  generate Common Name field to be specified in the generated child certificate
        child_cert_name = f"{cert_name} - CERT"

        # generate a CSR to sign the child certificate
        child_csr = make_csr(child_key, child_cert_name).decode()

        # sign the CSR by the intermediate CA (make sure inter. cert. is available via AIA)
        child_ca = sign_csr(child_csr, inter_ca, inter_key, 1096,
                            f"""
                                {get_aia_cnf_line("http://localhost:5000/static/openssl/inter.crt")}
                                """).decode()

        # generate a path of a file in which the child cert. will be stored
        child_ca_file = get_path(f"{cert_name}.crt")
        # store the generated child cert.
        puts(child_ca, child_ca_file)

        # store the generated private key used to generate the child. cert
        puts(child_key, get_path(f"{cert_name}.key"))

        #####################################
        # TEST SECTION

        # read content of the root CA certificate
        root_cert_content = subprocess.check_output(["openssl", "x509", "-noout", "-text", "-in", "-"],
                                                    input=bytes(read_file(get_path("root.crt")),
                                                                encoding="utf-8")).decode()

        # assert the common name of the certificate and that the certificate represents a CA
        assert f"Subject: CN = ROOT CA" in root_cert_content
        assert "CA:TRUE" in root_cert_content

        # read content of the intermediate CA certificate
        inter_cert_content = subprocess.check_output(["openssl", "x509", "-noout", "-text", "-in", "-"],
                                                     input=bytes(inter_ca, encoding="utf-8")).decode()

        # assert that the issuer of the certificate is the root CA
        assert "Issuer: CN = ROOT CA" in inter_cert_content
        # assert the common name
        assert f"Subject: CN = INTER CA" in inter_cert_content
        # assert that the certificate contains a reference to the root CA's certificate via AIA
        assert "CA Issuers - URI:http://localhost:5000/static/openssl/root.crt" in inter_cert_content
        # ensure that the generated certificate is indeed a CA
        assert "CA:TRUE" in inter_cert_content

        # read content of the child certificate
        child_cert_content = subprocess.check_output(
            ["openssl", "x509", "-noout", "-text", "-in", child_ca_file]).decode()

        # assert that the issuer is the intermediate CA
        assert "Issuer: CN = INTER CA" in child_cert_content
        # assert that the common name is the one passed via arguments
        assert f"Subject: CN = {child_cert_name}" in child_cert_content
        # assert that the certificate contains a reference to the intermediate CA's certificate via AIA
        assert "CA Issuers - URI:http://localhost:5000/static/openssl/inter.crt" in child_cert_content
        # assert that the generated certificate is not a CA
        assert "CA:TRUE" not in child_cert_content

        # assert that the chain of trust is valid (root -> intermediate -> child)
        assert subprocess.check_output(
            ["openssl", "verify", "-CAfile", get_path("root.crt"), "-untrusted", get_path("inter.crt"), child_ca_file])

        print(f"TESTS PASSED!!! \n{cert_name} certificate created successfully")