gistfile1.txt

from os.path import join as join_path
from os import listdir
from os import mkdir
from os.path import exists as path_exists
from os.path import expanduser

from OpenSSL.crypto import X509Extension
from OpenSSL.crypto import X509
from OpenSSL.crypto import dump_privatekey
from OpenSSL.crypto import dump_certificate
from OpenSSL.crypto import load_certificate
from OpenSSL.crypto import load_privatekey
from OpenSSL.crypto import PKey
from OpenSSL.crypto import TYPE_RSA
from OpenSSL.crypto import X509Req
from OpenSSL.SSL import FILETYPE_PEM


class CertificateSpoofer(object):

    def __init__(self):
        self.ca_path = expanduser('~/.proxygen')
        self.ca_fn = join_path(self.ca_path, 'ca.pem')
        self.cache_dir = join_path(self.ca_path, 'spoofed_certs')
        self._serial = len(listdir(self.cache_dir))

        if not path_exists(self.cache_dir):
            mkdir(self.cache_dir)
        if not path_exists(self.ca_fn):
            self._generate_ca()
        else:
            cert = open(self.ca_fn).read()
            self.cert = load_certificate(FILETYPE_PEM, cert)
            self.key = load_privatekey(FILETYPE_PEM, cert)

    def _generate_ca(self):
        # Generate key
        self.key = PKey()
        self.key.generate_key(TYPE_RSA, 2048)

        # Generate certificate
        self.cert = X509()
        self.cert.set_version(3)
        self.cert.set_serial_number(1)
        self.cert.get_subject().CN = 'ca.mitm.com'
        self.cert.gmtime_adj_notBefore(0)
        self.cert.gmtime_adj_notAfter(315360000)
        self.cert.set_issuer(self.cert.get_subject())
        self.cert.set_pubkey(self.key)
        self.cert.add_extensions([
            X509Extension("basicConstraints", True, "CA:TRUE, pathlen:0"),
            X509Extension("keyUsage", True, "keyCertSign, cRLSign"),
            X509Extension(
                "subjectKeyIdentifier", False, "hash", subject=self.cert),
        ])
        self.cert.sign(self.key, "sha1")

        with open(self.ca_file, 'wb+') as f:
            f.write(dump_privatekey(FILETYPE_PEM, self.key))
            f.write(dump_certificate(FILETYPE_PEM, self.cert))

    def __getitem__(self, cn):
        cnp = join_path(self.cache_dir, 'spoofed_%s.pem' % cn)
        if not path_exists(cnp):
            self._serial += 1
            # create certificate
            key = PKey()
            key.generate_key(TYPE_RSA, 2048)

            # Generate CSR
            req = X509Req()
            req.get_subject().CN = cn
            req.set_pubkey(key)
            req.sign(key, 'sha1')

            # Sign CSR
            cert = X509()
            cert.set_subject(req.get_subject())
            cert.set_serial_number(self.serial)
            cert.gmtime_adj_notBefore(0)
            cert.gmtime_adj_notAfter(31536000)
            cert.set_issuer(self.cert.get_subject())
            cert.set_pubkey(req.get_pubkey())
            cert.sign(self.key, 'sha1')

            with open(cnp, 'wb+') as f:
                f.write(dump_privatekey(FILETYPE_PEM, key))
                f.write(dump_certificate(FILETYPE_PEM, cert))

        return cnp

    @property
    def serial(self):
        return self._serial