Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-23 08:11:00 +00:00
Code Issues Projects Releases Wiki Activity

CertStore: add support for cert chains

Browse Source
This commit is contained in:
Maximilian Hils 2014-10-08 20:46:30 +02:00
parent 274688172d
commit fdb6f5552d
4 changed files with 55 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
netlib/certutils.py
View File

@ -113,13 +113,21 @@ def dummy_cert(privkey, cacert, commonname, sans):
# return current.value
class CertStoreEntry(object):
def __init__(self, cert, pkey=None, chain_file=None):
self.cert = cert
self.pkey = pkey
self.chain_file = chain_file
class CertStore:
"""
Implements an in-memory certificate store.
"""
def __init__(self, privkey, cacert, dhparams=None):
self.privkey, self.cacert = privkey, cacert
def __init__(self, default_pkey, default_ca, default_chain_file, dhparams=None):
self.default_pkey = default_pkey
self.default_ca = default_ca
self.default_chain_file = default_chain_file
self.dhparams = dhparams
self.certs = dict()
@ -142,21 +150,21 @@ class CertStore:
return dh
@classmethod
def from_store(klass, path, basename):
p = os.path.join(path, basename + "-ca.pem")
if not os.path.exists(p):
key, ca = klass.create_store(path, basename)
def from_store(cls, path, basename):
ca_path = os.path.join(path, basename + "-ca.pem")
if not os.path.exists(ca_path):
key, ca = cls.create_store(path, basename)
else:
p = os.path.join(path, basename + "-ca.pem")
raw = file(p, "rb").read()
with open(ca_path, "rb") as f:
raw = f.read()
ca = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, raw)
key = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, raw)
dhp = os.path.join(path, basename + "-dhparam.pem")
dh = klass.load_dhparam(dhp)
return klass(key, ca, dh)
dh_path = os.path.join(path, basename + "-dhparam.pem")
dh = cls.load_dhparam(dh_path)
return cls(key, ca, ca_path, dh)
@classmethod
def create_store(klass, path, basename, o=None, cn=None, expiry=DEFAULT_EXP):
def create_store(cls, path, basename, o=None, cn=None, expiry=DEFAULT_EXP):
if not os.path.exists(path):
os.makedirs(path)
@ -194,25 +202,29 @@ class CertStore:
return key, ca
def add_cert_file(self, spec, path):
raw = file(path, "rb").read()
cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, raw)
with open(path, "rb") as f:
raw = f.read()
cert = SSLCert(OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, raw))
try:
privkey = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, raw)
pkey = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, raw)
except Exception:
privkey = None
self.add_cert(SSLCert(cert), privkey, spec)
pkey = None
self.add_cert(
CertStoreEntry(cert, pkey, path),
spec
)
def add_cert(self, cert, privkey, *names):
def add_cert(self, entry, *names):
"""
Adds a cert to the certstore. We register the CN in the cert plus
any SANs, and also the list of names provided as an argument.
"""
if cert.cn:
self.certs[cert.cn] = (cert, privkey)
for i in cert.altnames:
self.certs[i] = (cert, privkey)
if entry.cert.cn:
self.certs[entry.cert.cn] = entry
for i in entry.cert.altnames:
self.certs[i] = entry
for i in names:
self.certs[i] = (cert, privkey)
self.certs[i] = entry
@staticmethod
def asterisk_forms(dn):
@ -246,17 +258,17 @@ class CertStore:
name = next(itertools.ifilter(lambda key: key in self.certs, potential_keys), None)
if name:
c = self.certs[name]
entry = self.certs[name]
else:
c = dummy_cert(self.privkey, self.cacert, commonname, sans), None
self.certs[(commonname, tuple(sans))] = c
entry = CertStoreEntry(cert=dummy_cert(self.default_pkey, self.default_ca, commonname, sans))
self.certs[(commonname, tuple(sans))] = entry
return c[0], (c[1] or self.privkey)
return entry.cert, (entry.pkey or self.default_pkey), (entry.chain_file or self.default_chain_file)
def gen_pkey(self, cert):
from . import certffi
certffi.set_flags(self.privkey, 1)
return self.privkey
certffi.set_flags(self.default_pkey, 1)
return self.default_pkey
class _GeneralName(univ.Choice):

6
netlib/tcp.py
View File

@ -345,7 +345,7 @@ class BaseHandler(_Connection):
def _create_ssl_context(self, cert, key, method=SSLv23_METHOD, options=None,
handle_sni=None, request_client_cert=None, cipher_list=None,
dhparams=None, ca_file=None):
dhparams=None, chain_file=None):
"""
cert: A certutils.SSLCert object.
@ -377,8 +377,8 @@ class BaseHandler(_Connection):
ctx = SSL.Context(method)
if not options is None:
ctx.set_options(options)
if ca_file:
ctx.load_verify_locations(ca_file)
if chain_file:
ctx.load_verify_locations(chain_file)
if cipher_list:
try:
ctx.set_cipher_list(cipher_list)

14
test/test_certutils.py
View File

@ -42,7 +42,7 @@ class TestCertStore:
ca2 = certutils.CertStore.from_store(d, "test")
assert ca2.get_cert("foo", [])
assert ca.cacert.get_serial_number() == ca2.cacert.get_serial_number()
assert ca.default_ca.get_serial_number() == ca2.default_ca.get_serial_number()
def test_create_tmp(self):
with tutils.tmpdir() as d:
@ -52,7 +52,7 @@ class TestCertStore:
assert ca.get_cert("*.foo.com", [])
r = ca.get_cert("*.foo.com", [])
assert r[1] == ca.privkey
assert r[1] == ca.default_pkey
def test_add_cert(self):
with tutils.tmpdir() as d:
@ -71,14 +71,14 @@ class TestCertStore:
with tutils.tmpdir() as d:
ca = certutils.CertStore.from_store(d, "test")
_ = ca.get_cert("foo.com", ["*.bar.com"])
cert, key = ca.get_cert("foo.bar.com", ["*.baz.com"])
cert, key, chain_file = ca.get_cert("foo.bar.com", ["*.baz.com"])
assert "*.baz.com" in cert.altnames
def test_overrides(self):
with tutils.tmpdir() as d:
ca1 = certutils.CertStore.from_store(os.path.join(d, "ca1"), "test")
ca2 = certutils.CertStore.from_store(os.path.join(d, "ca2"), "test")
assert not ca1.cacert.get_serial_number() == ca2.cacert.get_serial_number()
assert not ca1.default_ca.get_serial_number() == ca2.default_ca.get_serial_number()
dc = ca2.get_cert("foo.com", [])
dcp = os.path.join(d, "dc")
@ -98,7 +98,7 @@ class TestCertStore:
cert = ca1.get_cert("foo.com", [])
assert certffi.get_flags(ca2.gen_pkey(cert[0])) == 1
finally:
certffi.set_flags(ca2.privkey, 0)
certffi.set_flags(ca2.default_pkey, 0)
class TestDummyCert:
@ -106,8 +106,8 @@ class TestDummyCert:
with tutils.tmpdir() as d:
ca = certutils.CertStore.from_store(d, "test")
r = certutils.dummy_cert(
ca.privkey,
ca.cacert,
ca.default_pkey,
ca.default_ca,
"foo.com",
["one.com", "two.com", "*.three.com"]
)

8
test/test_tcp.py
View File

@ -393,7 +393,7 @@ class TestPrivkeyGen(test.ServerTestBase):
with tutils.tmpdir() as d:
ca1 = certutils.CertStore.from_store(d, "test2")
ca2 = certutils.CertStore.from_store(d, "test3")
cert, _ = ca1.get_cert("foo.com", [])
cert, _, _ = ca1.get_cert("foo.com", [])
key = ca2.gen_pkey(cert)
self.convert_to_ssl(cert, key)
@ -409,9 +409,9 @@ class TestPrivkeyGenNoFlags(test.ServerTestBase):
with tutils.tmpdir() as d:
ca1 = certutils.CertStore.from_store(d, "test2")
ca2 = certutils.CertStore.from_store(d, "test3")
cert, _ = ca1.get_cert("foo.com", [])
certffi.set_flags(ca2.privkey, 0)
self.convert_to_ssl(cert, ca2.privkey)
cert, _, _ = ca1.get_cert("foo.com", [])
certffi.set_flags(ca2.default_pkey, 0)
self.convert_to_ssl(cert, ca2.default_pkey)
def test_privkey(self):
c = tcp.TCPClient(("127.0.0.1", self.port))