<commit_before>import functools
import os
import shutil
import subprocess
import tempfile
from datetime import datetime

class OpenSSL(object):
    def __init__(self, logger, binary, base_path, conf_path, hosts, duration,
                 base_conf_path=None):
        """Context manager for interacting with OpenSSL.
        Creates a config file for the duration of the context.

        :param logger: stdlib logger or python structured logger
        :param binary: path to openssl binary
        :param base_path: path to directory for storing certificates
        :param conf_path: path for configuration file storing configuration data
        :param hosts: list of hosts to include in configuration (or None if not
                      generating host certificates)
        :param duration: Certificate duration in days"""

        self.base_path = base_path
        self.binary = binary
        self.conf_path = conf_path
        self.base_conf_path = base_conf_path
        self.logger = logger
        self.proc = None
        self.cmd = []
        self.hosts = hosts
        self.duration = duration

    def __enter__(self):
        with open(self.conf_path, "w") as f:
            f.write(get_config(self.base_path, self.hosts, self.duration))
        return self

    def __exit__(self, *args, **kwargs):
        os.unlink(self.conf_path)

    def log(self, line):
        if hasattr(self.logger, "process_output"):
            self.logger.process_output(self.proc.pid if self.proc is not None else None,
                                       line.decode("utf8", "replace"),
                                       command=" ".join(self.cmd))
        else:
            self.logger.debug(line)

    def __call__(self, cmd, *args, **kwargs):
        """Run a command using OpenSSL in the current context.

        :param cmd: The openssl subcommand to run
        :param *args: Additional arguments to pass to the command
        """
        self.cmd = [self.binary, cmd]
        if cmd != "x509":
            self.cmd += ["-config", self.conf_path]
        self.cmd += list(args)

        env = os.environ.copy()
        if self.base_conf_path is not None:
            env["OPENSSL_CONF"] = self.base_conf_path.encode("utf8")

        self.proc = subprocess.Popen(self.cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                     env=env)
        stdout, stderr = self.proc.communicate()
        self.log(stdout)
        if self.proc.returncode != 0:
            raise subprocess.CalledProcessError(self.proc.returncode, self.cmd,
                                                output=stdout)

        self.cmd = []
        self.proc = None
        return stdout


def make_subject(common_name,
                 country=None,
                 state=None,
                 locality=None,
                 organization=None,
                 organization_unit=None):
    args = [("country", "C"),
            ("state", "ST"),
            ("locality", "L"),
            ("organization", "O"),
            ("organization_unit", "OU"),
            ("common_name", "CN")]

    rv = []

    for var, key in args:
        value = locals()[var]
        if value is not None:
            rv.append("/%s=%s" % (key, value.replace("/", "\\/")))

    return "".join(rv)

def make_alt_names(hosts):
    rv = []
    for name in hosts:
        rv.append("DNS:%s" % name)
    return ",".join(rv)

def get_config(root_dir, hosts, duration=30):
    if hosts is None:
        san_line = ""
    else:
        san_line = "subjectAltName = %s" % make_alt_names(hosts)

    if os.path.sep == "\\":
        # This seems to be needed for the Shining Light OpenSSL on
        # Windows, at least.
        root_dir = root_dir.replace("\\", "\\\\")

    rv = """[ ca ]
default_ca = CA_default

[ CA_default ]
dir = %(root_dir)s
certs = $dir
new_certs_dir = $certs
crl_dir = $dir%(sep)scrl
database = $dir%(sep)sindex.txt
private_key = $dir%(sep)scakey.pem
certificate = $dir%(sep)scacert.pem
serial = $dir%(sep)sserial
crldir = $dir%(sep)scrl
crlnumber = $dir%(sep)scrlnumber
crl = $crldir%(sep)scrl.pem
RANDFILE = $dir%(sep)sprivate%(sep)s.rand
x509_extensions = usr_cert
name_opt        = ca_default
cert_opt        = ca_default
default_days = %(duration)d
default_crl_days = %(duration)d
default_md = sha256
preserve = no
policy = policy_anything
copy_extensions = copy

[ policy_anything ]
countryName = optional
stateOrProvinceName = optional
localityName = optional
organizationName = optional
organizationalUnitName = optional
commonName = supplied
emailAddress = optional

[ req ]
default_bits = 2048
default_keyfile  = privkey.pem
distinguished_name = req_distinguished_name
attributes = req_attributes
x509_extensions = v3_ca

# Passwords for private keys if not present they will be prompted for
# input_password = <PASSWORD>
# output_password = <PASSWORD>
string_mask = utf8only
req_extensions = v3_req

[ req_distinguished_name ]
countryName = Country Name (2 letter code)
countryName_default = AU
countryName_min = 2
countryName_max = 2
stateOrProvinceName = State or Province Name (full name)
stateOrProvinceName_default =
localityName = Locality Name (eg, city)
0.organizationName = Organization Name
0.organizationName_default = Web Platform Tests
organizationalUnitName = Organizational Unit Name (eg, section)
#organizationalUnitName_default =
commonName = Common Name (e.g. server FQDN or YOUR name)
commonName_max = 64
emailAddress = Email Address
emailAddress_max = 64

[ req_attributes ]

[ usr_cert ]
basicConstraints=CA:false
subjectKeyIdentifier=hash
authorityKeyIdentifier=keyid,issuer

[ v3_req ]
basicConstraints = CA:FALSE
keyUsage = nonRepudiation, digitalSignature, keyEncipherment
extendedKeyUsage = serverAuth
%(san_line)s

[ v3_ca ]
basicConstraints = CA:true
subjectKeyIdentifier=hash
authorityKeyIdentifier=keyid:always,issuer:always
keyUsage = keyCertSign
""" % {"root_dir": root_dir,
       "san_line": san_line,
       "duration": duration,
       "sep": os.path.sep}

    return rv

class OpenSSLEnvironment(object):
    ssl_enabled = True

    def __init__(self, logger, openssl_binary="openssl", base_path=None,
                 password="<PASSWORD>", force_regenerate=False,
                 duration=30, base_conf_path=None):
        """SSL environment that creates a local CA and host certificate using OpenSSL.

        By default this will look in base_path for existing certificates that are still
        valid and only create new certificates if there aren't any. This behaviour can
        be adjusted using the force_regenerate option.

        :param logger: a stdlib logging compatible logger or mozlog structured logger
        :param openssl_binary: Path to the OpenSSL binary
        :param base_path: Path in which certificates will be stored. If None, a temporary
                          directory will be used and removed when the server shuts down
        :param password: <PASSWORD>
        :param force_regenerate: Always create a new certificate even if one already exists.
        """
        self.logger = logger

        self.temporary = False
        if base_path is None:
            base_path = tempfile.mkdtemp()
            self.temporary = True

        self.base_path = os.path.abspath(base_path)
        self.password = password<commit_msg>Ensure that all path separators in openssl config on windows are written \\ not \<commit_after>import functools
import os
import shutil
import subprocess
import tempfile
from datetime import datetime

class OpenSSL(object):
    def __init__(self, logger, binary, base_path, conf_path, hosts, duration,
                 base_conf_path=None):
        """Context manager for interacting with OpenSSL.
        Creates a config file for the duration of the context.

        :param logger: stdlib logger or python structured logger
        :param binary: path to openssl binary
        :param base_path: path to directory for storing certificates
        :param conf_path: path for configuration file storing configuration data
        :param hosts: list of hosts to include in configuration (or None if not
                      generating host certificates)
        :param duration: Certificate duration in days"""

        self.base_path = base_path
        self.binary = binary
        self.conf_path = conf_path
        self.base_conf_path = base_conf_path
        self.logger = logger
        self.proc = None
        self.cmd = []
        self.hosts = hosts
        self.duration = duration

    def __enter__(self):
        with open(self.conf_path, "w") as f:
            f.write(get_config(self.base_path, self.hosts, self.duration))
        return self

    def __exit__(self, *args, **kwargs):
        os.unlink(self.conf_path)

    def log(self, line):
        if hasattr(self.logger, "process_output"):
            self.logger.process_output(self.proc.pid if self.proc is not None else None,
                                       line.decode("utf8", "replace"),
                                       command=" ".join(self.cmd))
        else:
            self.logger.debug(line)

    def __call__(self, cmd, *args, **kwargs):
        """Run a command using OpenSSL in the current context.

        :param cmd: The openssl subcommand to run
        :param *args: Additional arguments to pass to the command
        """
        self.cmd = [self.binary, cmd]
        if cmd != "x509":
            self.cmd += ["-config", self.conf_path]
        self.cmd += list(args)

        env = os.environ.copy()
        if self.base_conf_path is not None:
            env["OPENSSL_CONF"] = self.base_conf_path.encode("utf8")

        self.proc = subprocess.Popen(self.cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                     env=env)
        stdout, stderr = self.proc.communicate()
        self.log(stdout)
        if self.proc.returncode != 0:
            raise subprocess.CalledProcessError(self.proc.returncode, self.cmd,
                                                output=stdout)

        self.cmd = []
        self.proc = None
        return stdout


def make_subject(common_name,
                 country=None,
                 state=None,
                 locality=None,
                 organization=None,
                 organization_unit=None):
    args = [("country", "C"),
            ("state", "ST"),
            ("locality", "L"),
            ("organization", "O"),
            ("organization_unit", "OU"),
            ("common_name", "CN")]

    rv = []

    for var, key in args:
        value = locals()[var]
        if value is not None:
            rv.append("/%s=%s" % (key, value.replace("/", "\\/")))

    return "".join(rv)

def make_alt_names(hosts):
    rv = []
    for name in hosts:
        rv.append("DNS:%s" % name)
    return ",".join(rv)

def get_config(root_dir, hosts, duration=30):
    if hosts is None:
        san_line = ""
    else:
        san_line = "subjectAltName = %s" % make_alt_names(hosts)

    if os.path.sep == "\\":
        # This seems to be needed for the Shining Light OpenSSL on
        # Windows, at least.
        root_dir = root_dir.replace("\\", "\\\\")

    rv = """[ ca ]
default_ca = CA_default

[ CA_default ]
dir = %(root_dir)s
certs = $dir
new_certs_dir = $certs
crl_dir = $dir%(sep)scrl
database = $dir%(sep)sindex.txt
private_key = $dir%(sep)scakey.pem
certificate = $dir%(sep)scacert.pem
serial = $dir%(sep)sserial
crldir = $dir%(sep)scrl
crlnumber = $dir%(sep)scrlnumber
crl = $crldir%(sep)scrl.pem
RANDFILE = $dir%(sep)sprivate%(sep)s.rand
x509_extensions = usr_cert
name_opt        = ca_default
cert_opt        = ca_default
default_days = %(duration)d
default_crl_days = %(duration)d
default_md = sha256
preserve = no
policy = policy_anything
copy_extensions = copy

[ policy_anything ]
countryName = optional
stateOrProvinceName = optional
localityName = optional
organizationName = optional
organizationalUnitName = optional
commonName = supplied
emailAddress = optional

[ req ]
default_bits = 2048
default_keyfile  = privkey.pem
distinguished_name = req_distinguished_name
attributes = req_attributes
x509_extensions = v3_ca

# Passwords for private keys if not present they will be prompted for
# input_password = <PASSWORD>
# output_password = <PASSWORD>
string_mask = utf8only
req_extensions = v3_req

[ req_distinguished_name ]
countryName = Country Name (2 letter code)
countryName_default = AU
countryName_min = 2
countryName_max = 2
stateOrProvinceName = State or Province Name (full name)
stateOrProvinceName_default =
localityName = Locality Name (eg, city)
0.organizationName = Organization Name
0.organizationName_default = Web Platform Tests
organizationalUnitName = Organizational Unit Name (eg, section)
#organizationalUnitName_default =
commonName = Common Name (e.g. server FQDN or YOUR name)
commonName_max = 64
emailAddress = Email Address
emailAddress_max = 64

[ req_attributes ]

[ usr_cert ]
basicConstraints=CA:false
subjectKeyIdentifier=hash
authorityKeyIdentifier=keyid,issuer

[ v3_req ]
basicConstraints = CA:FALSE
keyUsage = nonRepudiation, digitalSignature, keyEncipherment
extendedKeyUsage = serverAuth
%(san_line)s

[ v3_ca ]
basicConstraints = CA:true
subjectKeyIdentifier=hash
authorityKeyIdentifier=keyid:always,issuer:always
keyUsage = keyCertSign
""" % {"root_dir": root_dir,
       "san_line": san_line,
       "duration": duration,
       "sep": os.path.sep.replace("\\", "\\\\")}

    return rv

class OpenSSLEnvironment(object):
    ssl_enabled = True

    def __init__(self, logger, openssl_binary="openssl", base_path=None,
                 password="<PASSWORD>", force_regenerate=False,
                 duration=30, base_conf_path=None):
        """SSL environment that creates a local CA and host certificate using OpenSSL.

        By default this will look in base_path for existing certificates that are still
        valid and only create new certificates if there aren't any. This behaviour can
        be adjusted using the force_regenerate option.

        :param logger: a stdlib logging compatible logger or mozlog structured logger
        :param openssl_binary: Path to the OpenSSL binary
        :param base_path: Path in which certificates will be stored. If None, a temporary
                          directory will be used and removed when the server shuts down
        :param password: <PASSWORD>
        :param force_regenerate: Always create a new certificate even if one already exists.
        """
        self.logger = logger

        self.temporary = False
        if base_path is None:
            base_path = tempfile.mkdtemp()
            self.temporary = True

        self.base_path = os.path.abspath(base_path)
        self.password = password
{'fraction_non_alphanumeric': 0.09341552635144289, 'fraction_numerical': 0.00501287088470397, 'mean_word_length': 2.670561909497762, 'pattern_counts': {'":': 12, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1833506', 'n_tokens_mistral': 4363, 'n_tokens_neox': 4179, 'n_words': 1234}