<commit_before>"""Routines to generate dummy certificates."""

import os
import shutil
import tempfile
import time

openssl_exception = None
try:
  from OpenSSL import crypto
except ImportError, e:
  openssl_exception = e


def generate_dummy_ca(path):
  """Generates dummy certificate authority and writes it to file."""
  if openssl_exception:
    raise openssl_exception
  dirname = os.path.dirname(path)
  if not os.path.exists(dirname):
    os.makedirs(dirname)
  if path.endswith('.pem'):
    basename, _ = os.path.splitext(path)
    basename = os.path.basename(basename)
  else:
    basename = os.path.basename(path)

  key = crypto.PKey()
  key.generate_key(crypto.TYPE_RSA, 1024)

  ca = crypto.X509()
  ca.set_serial_number(int(time.time()*10000))
  ca.set_version(2)
  ca.get_subject().CN = 'sslproxy'
  ca.get_subject().O = 'sslproxy'
  ca.gmtime_adj_notBefore(-60 * 60 * 24 * 365 * 2)
  ca.gmtime_adj_notAfter(60 * 60 * 24 * 365 * 2)
  ca.set_issuer(ca.get_subject())
  ca.set_pubkey(key)
  ca.add_extensions([
      crypto.X509Extension('basicConstraints', True, 'CA:TRUE'),
      crypto.X509Extension('nsCertType', True, 'sslCA'),
      crypto.X509Extension('extendedKeyUsage', True,
                           ('serverAuth,clientAuth,emailProtection,'
                            'timeStamping,msCodeInd,msCodeCom,msCTLSign,'
                            'msSGC,msEFS,nsSGC')),
      crypto.X509Extension('keyUsage', False, 'keyCertSign, cRLSign'),
      crypto.X509Extension('subjectKeyIdentifier', False, 'hash', subject=ca),
      ])
  ca.sign(key, 'sha1')

  # Dump the CA plus private key
  with open(path, 'w') as f:
    f.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, key))
    f.write(crypto.dump_certificate(crypto.FILETYPE_PEM, ca))

  # Dump the certificate in PEM format
  with open(os.path.join(dirname, basename + '-cert.pem'), 'w') as f:
    f.write(crypto.dump_certificate(crypto.FILETYPE_PEM, ca))

  # Create a .cer file with the same contents for Android
  with open(os.path.join(dirname, basename + '-cert.cer'), 'w') as f:
    f.write(crypto.dump_certificate(crypto.FILETYPE_PEM, ca))

  # Dump the certificate in PKCS12 format for Windows devices
  with open(os.path.join(dirname, basename + '-cert.p12'), 'w') as f:
    p12 = crypto.PKCS12()
    p12.set_certificate(ca)
    p12.set_privatekey(key)
    f.write(p12.export())


def generate_dummy_cert(path, ca, commonname):
  """Generates a certificate for |commonname| signed by |ca| in |path|."""
  if openssl_exception:
    raise openssl_exception
  raw = open(ca, 'r').read()
  ca = crypto.load_certificate(crypto.FILETYPE_PEM, raw)
  key = crypto.load_privatekey(crypto.FILETYPE_PEM, raw)

  req = crypto.X509Req()
  subj = req.get_subject()
  subj.CN = commonname
  req.set_pubkey(ca.get_pubkey())
  req.sign(key, 'sha1')

  cert = crypto.X509()
  cert.gmtime_adj_notBefore(-60 * 60)
  cert.gmtime_adj_notAfter(60 * 60 * 24 * 30)
  cert.set_issuer(ca.get_subject())
  cert.set_subject(req.get_subject())
  cert.set_serial_number(int(time.time()*10000))
  cert.set_pubkey(req.get_pubkey())
  cert.sign(key, 'sha1')

  with open(path, 'w') as f:
    f.write(crypto.dump_certificate(crypto.FILETYPE_PEM, cert))


class CertStore(object):
  """Implements an on-disk certificate store."""

  def __init__(self, cacert, certdir=None):
    if not os.path.exists(cacert):
      generate_dummy_ca(cacert)
    self.cacert = cacert
    if certdir:
      self.remove = False
      self.certdir = certdir
    else:
      self.remove = True
      self.certdir = tempfile.mkdtemp(prefix='certstore')

  def check_domain(self, commonname):
    """Checks for valid domain."""
    try:
      commonname.decode('idna')
      commonname.decode('ascii')
    except:
      return False
    if '..' in commonname:
      return False
    if '/' in commonname:
      return False
    return True

  def get_cert(self, commonname):
    """Returns the path to the certificate for |commonname|."""
    if not self.check_domain(commonname):
      return None
    path = os.path.join(self.certdir, commonname + '.pem')
    if not os.path.exists(path):
      generate_dummy_cert(path, self.cacert, commonname)
    return path

  def cleanup(self):
    if self.remove:
      shutil.rmtree(self.certdir)
<commit_msg>fixing certutils<commit_after>"""Routines to generate dummy certificates."""

import os
import shutil
import tempfile
import time

openssl_exception = None
try:
  from OpenSSL import crypto
except ImportError, e:
  openssl_exception = e


def generate_dummy_ca():
  """Generates dummy certificate authority.
  """ 
  if openssl_exception:
    raise openssl_exception

  key = crypto.PKey()
  key.generate_key(crypto.TYPE_RSA, 1024)

  ca = crypto.X509()
  ca.set_serial_number(int(time.time()*10000))
  ca.set_version(2)
  ca.get_subject().CN = 'sslproxy'
  ca.get_subject().O = 'sslproxy'
  ca.gmtime_adj_notBefore(-60 * 60 * 24 * 365 * 2)
  ca.gmtime_adj_notAfter(60 * 60 * 24 * 365 * 2)
  ca.set_issuer(ca.get_subject())
  ca.set_pubkey(key)
  ca.add_extensions([
      crypto.X509Extension('basicConstraints', True, 'CA:TRUE'),
      crypto.X509Extension('nsCertType', True, 'sslCA'),
      crypto.X509Extension('extendedKeyUsage', True,
                           ('serverAuth,clientAuth,emailProtection,'
                            'timeStamping,msCodeInd,msCodeCom,msCTLSign,'
                            'msSGC,msEFS,nsSGC')),
      crypto.X509Extension('keyUsage', False, 'keyCertSign, cRLSign'),
      crypto.X509Extension('subjectKeyIdentifier', False, 'hash', subject=ca),
      ])
  ca.sign(key, 'sha1')
  return ca, key

def write_dummy_ca(cert_path, ca, key):
  """ Writes four certificate files. For example, if cert_path is "mycert.pem":
      mycert.pem - CA plus private key
      mycert-cert.pem - CA in PEM format
      mycert-cert.cer - CA for Android
      mycert-cert.p12 - CA in PKCS12 format for Windows devices
  Args:
    cert_path: path string such as "mycert.pem"
    ca: crypto X509 generated certificate
    key: crypto X509 generated private key
  """ 
  dirname = os.path.dirname(cert_path)
  if dirname and not os.path.exists(dirname):
    os.makedirs(dirname)

  root_path = os.path.splitext(cert_path)[0]
  pem_path = root_path + '-cert.pem'
  android_cer_path = root_path + '-cert.cer'
  windows_p12_path = root_path + '-cert.p12'

  pem_key = crypto.dump_privatekey(crypto.FILETYPE_PEM, key)
  pem_ca = crypto.dump_certificate(crypto.FILETYPE_PEM, ca)

  # Dump the CA plus private key
  with open(cert_path, 'w') as f:
    f.write(pem_key)
    f.write(pem_ca)

  # Dump the certificate in PEM format
  with open(pem_path , 'w') as f:
    f.write(pem_ca)

  # Create a .cer file with the same contents for Android
  with open(android_cer_path , 'w') as f:
    f.write(pem_ca)

  # Dump the certificate in PKCS12 format for Windows devices
  with open(windows_p12_path , 'w') as f:
    p12 = crypto.PKCS12()
    p12.set_certificate(ca)
    p12.set_privatekey(key)
    f.write(p12.export())

def generate_dummy_cert(path, ca, common_name):
  """Generates a certificate for |common_name| signed by |ca| in |path|."""
  if openssl_exception:
    raise openssl_exception
  raw = open(ca, 'r').read()
  ca = crypto.load_certificate(crypto.FILETYPE_PEM, raw)
  key = crypto.load_privatekey(crypto.FILETYPE_PEM, raw)

  req = crypto.X509Req()
  subj = req.get_subject()
  subj.CN = common_name
  req.set_pubkey(ca.get_pubkey())
  req.sign(key, 'sha1')

  cert = crypto.X509()
  cert.gmtime_adj_notBefore(-60 * 60)
  cert.gmtime_adj_notAfter(60 * 60 * 24 * 30)
  cert.set_issuer(ca.get_subject())
  cert.set_subject(req.get_subject())
  cert.set_serial_number(int(time.time()*10000))
  cert.set_pubkey(req.get_pubkey())
  cert.sign(key, 'sha1')

  with open(path, 'w') as f:
    f.write(crypto.dump_certificate(crypto.FILETYPE_PEM, cert))


class CertStore(object):
  """Implements an on-disk certificate store."""

  def __init__(self, ca_cert, cert_dir=None):
    if not os.path.exists(ca_cert):
      ca, key = generate_dummy_ca()
      write_dummy_ca(ca_cert, ca, key)
    self.ca_cert = ca_cert
    if cert_dir:
      self.is_temporary_cert_store = False
      self.cert_dir = cert_dir
    else:
      self.is_temporary_cert_store = True
      self.cert_dir = tempfile.mkdtemp(prefix='certstore')

  def is_valid_domain(self, common_name):
    """Checks for valid domain."""
    try:
      common_name.decode('idna')
      common_name.decode('ascii')
    except:
      return False
    return ('..' not in common_name and '/' not in common_name)

  def get_cert(self, common_name):
    """Returns the path to the certificate for |common_name|."""
    if not self.is_valid_domain(common_name):
      return None
    path = os.path.join(self.cert_dir, common_name + '.pem')
    if not os.path.exists(path):
      generate_dummy_cert(path, self.ca_cert, common_name)
    return path

  def cleanup(self):
    if self.is_temporary_cert_store:
      shutil.rmtree(self.cert_dir)

{'fraction_non_alphanumeric': 0.12469410456062291, 'fraction_numerical': 0.02124582869855395, 'mean_word_length': 3.5685975609756095, 'pattern_counts': {'":': 1, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4685583', 'n_tokens_mistral': 3321, 'n_tokens_neox': 3124, 'n_words': 682}