diff options
| -rw-r--r-- | libmproxy/proxy.py | 8 | ||||
| -rw-r--r-- | libmproxy/resources/ca.cnf | 33 | ||||
| -rw-r--r-- | libmproxy/resources/cert.cnf | 34 | ||||
| -rw-r--r-- | libmproxy/utils.py | 225 | ||||
| -rwxr-xr-x | mitmdump | 2 | ||||
| -rwxr-xr-x | mitmplayback | 2 | ||||
| -rwxr-xr-x | mitmproxy | 2 | ||||
| -rwxr-xr-x | mitmrecord | 2 | ||||
| -rw-r--r-- | test/test_utils.py | 44 |
9 files changed, 196 insertions, 156 deletions
diff --git a/libmproxy/proxy.py b/libmproxy/proxy.py index 4ab19694..e3c692b9 100644 --- a/libmproxy/proxy.py +++ b/libmproxy/proxy.py @@ -550,7 +550,7 @@ class ProxyHandler(SocketServer.StreamRequestHandler): if config.certpath is not None: cert = config.certpath + "/" + host + ".pem" if not os.path.exists(cert) and config.cacert is not None: - utils.make_bogus_cert(cert, ca=config.cacert, commonName=host) + utils.dummy_cert(config.certpath, config.cacert, host) if os.path.exists(cert): return cert print >> sys.stderr, "WARNING: Certificate missing for %s:%d! (%s)\n" % (host, port, cert) @@ -679,9 +679,9 @@ class ProxyServer(ServerBase): def set_mqueue(self, q): self.masterq = q - def process_request(self, request, client_address): - return ServerBase.process_request(self, request, client_address) - def finish_request(self, request, client_address): self.RequestHandlerClass(request, client_address, self, self.masterq) + def shutdown(self): + ServerBase.shutdown(self) + diff --git a/libmproxy/resources/ca.cnf b/libmproxy/resources/ca.cnf new file mode 100644 index 00000000..e46bb08f --- /dev/null +++ b/libmproxy/resources/ca.cnf @@ -0,0 +1,33 @@ +[ req ] +prompt = no +distinguished_name = req_distinguished_name +x509_extensions = v3_ca +req_extensions = v3_ca_req + +[ req_distinguished_name ] +organizationName = mitmproxy +commonName = Dummy CA + +[ v3_ca ] +subjectKeyIdentifier=hash +authorityKeyIdentifier=keyid:always,issuer +basicConstraints = critical,CA:true +keyUsage = cRLSign, keyCertSign +nsCertType = sslCA + +[ v3_ca_req ] +basicConstraints = critical,CA:true +keyUsage = cRLSign, keyCertSign +nsCertType = sslCA + +[ v3_cert ] +basicConstraints = CA:false +keyUsage = nonRepudiation, digitalSignature, keyEncipherment +nsCertType = server +subjectKeyIdentifier=hash +authorityKeyIdentifier=keyid:always,issuer + +[ v3_cert_req ] +basicConstraints = CA:false +keyUsage = nonRepudiation, digitalSignature, keyEncipherment +nsCertType = server diff --git a/libmproxy/resources/cert.cnf b/libmproxy/resources/cert.cnf new file mode 100644 index 00000000..9afae09f --- /dev/null +++ b/libmproxy/resources/cert.cnf @@ -0,0 +1,34 @@ +[ req ] +prompt = no +distinguished_name = req_distinguished_name +x509_extensions = v3_cert +req_extensions = v3_cert_req + +[ req_distinguished_name ] +organizationName = mitmproxy +commonName = %(commonname)s + +[ v3_ca ] +subjectKeyIdentifier=hash +authorityKeyIdentifier=keyid:always,issuer +basicConstraints = critical,CA:true +keyUsage = cRLSign, keyCertSign +nsCertType = sslCA + +[ v3_ca_req ] +basicConstraints = critical,CA:true +keyUsage = cRLSign, keyCertSign +nsCertType = sslCA + +[ v3_cert ] +basicConstraints = CA:false +keyUsage = nonRepudiation, digitalSignature, keyEncipherment +nsCertType = server +subjectKeyIdentifier=hash +authorityKeyIdentifier=keyid:always,issuer + +[ v3_cert_req ] +basicConstraints = CA:false +keyUsage = nonRepudiation, digitalSignature, keyEncipherment +nsCertType = server + diff --git a/libmproxy/utils.py b/libmproxy/utils.py index c9be7483..51a8e871 100644 --- a/libmproxy/utils.py +++ b/libmproxy/utils.py @@ -12,7 +12,7 @@ # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. -import re, os, subprocess, datetime, textwrap, errno +import re, os, subprocess, datetime, textwrap, errno, sys import optparse def format_timestamp(s): @@ -314,167 +314,118 @@ class Data: data = Data(__name__) -def make_openssl_conf(path, countryName=None, stateOrProvinceName=None, localityName=None, organizationName=None, organizationalUnitName=None, commonName=None, emailAddress=None, ca=False): - cnf = open(path, "w") - cnf.write("[ req ]\n") - cnf.write("prompt = no\n") - cnf.write("distinguished_name = req_distinguished_name\n") - if ca: - cnf.write("x509_extensions = v3_ca # The extentions to add to the self signed cert\n") - cnf.write("\n") - cnf.write("[ req_distinguished_name ]\n") - if countryName is not None: - cnf.write("countryName = %s\n" % (countryName) ) - cnf.write("stateOrProvinceName = %s\n" % (stateOrProvinceName) ) - cnf.write("localityName = %s\n" % (localityName) ) - cnf.write("organizationName = %s\n" % (organizationName) ) - cnf.write("organizationalUnitName = %s\n" % (organizationalUnitName) ) - cnf.write("commonName = %s\n" % (commonName) ) - cnf.write("emailAddress = %s\n" % (emailAddress) ) - cnf.write("\n") - cnf.write("[ v3_ca ]\n") - cnf.write("subjectKeyIdentifier=hash\n") - cnf.write("authorityKeyIdentifier=keyid:always,issuer\n") - if ca: - cnf.write("basicConstraints = critical,CA:true\n") - cnf.write("keyUsage = cRLSign, keyCertSign\n") - #cnf.write("nsCertType = sslCA, emailCA\n") - #cnf.write("subjectAltName=email:copy\n") - #cnf.write("issuerAltName=issuer:copy\n") - -def make_bogus_cert(certpath, countryName=None, stateOrProvinceName=None, localityName=None, organizationName="mitmproxy", organizationalUnitName=None, commonName="Dummy Certificate", emailAddress=None, ca=None, newca=False): - # Generates a bogus certificate like so: - # openssl req -config template -x509 -nodes -days 9999 -newkey rsa:1024 \ - # -keyout cert.pem -out cert.pem - - (path, ext) = os.path.splitext(certpath) +def dummy_ca(path): + """ + Creates a dummy CA, and writes it to path. + + This function also creates the necessary directories if they don't exist. + + Returns True if operation succeeded, False if not. + """ d = os.path.dirname(path) if not os.path.exists(d): os.makedirs(d) - - cnf = open(path+".cnf", "w") - cnf.write("[ req ]\n") - cnf.write("prompt = no\n") - cnf.write("distinguished_name = req_distinguished_name\n") - if newca: - cnf.write("x509_extensions = v3_ca\n") - cnf.write("req_extensions = v3_ca_req\n") + cmd = [ + "openssl", + "req", + "-new", + "-x509", + "-config", data.path("resources/ca.cnf"), + "-nodes", + "-days", "9999", + "-out", path, + "-newkey", "rsa:1024", + "-keyout", path, + ] + ret = subprocess.call( + cmd, + stderr=subprocess.PIPE, + stdout=subprocess.PIPE, + stdin=subprocess.PIPE + ) + if ret: + return False else: - cnf.write("x509_extensions = v3_cert\n") - cnf.write("req_extensions = v3_cert_req\n") - cnf.write("\n") - cnf.write("[ req_distinguished_name ]\n") - if countryName is not None: - cnf.write("countryName = %s\n" % (countryName) ) - if stateOrProvinceName is not None: - cnf.write("stateOrProvinceName = %s\n" % (stateOrProvinceName) ) - if localityName is not None: - cnf.write("localityName = %s\n" % (localityName) ) - if organizationName is not None: - cnf.write("organizationName = %s\n" % (organizationName) ) - if organizationalUnitName is not None: - cnf.write("organizationalUnitName = %s\n" % (organizationalUnitName) ) - if commonName is not None: - cnf.write("commonName = %s\n" % (commonName) ) - if emailAddress is not None: - cnf.write("emailAddress = %s\n" % (emailAddress) ) - cnf.write("\n") - cnf.write("[ v3_ca ]\n") - cnf.write("subjectKeyIdentifier=hash\n") - cnf.write("authorityKeyIdentifier=keyid:always,issuer\n") - cnf.write("basicConstraints = critical,CA:true\n") - cnf.write("keyUsage = cRLSign, keyCertSign\n") - cnf.write("nsCertType = sslCA\n") - #cnf.write("subjectAltName=email:copy\n") - #cnf.write("issuerAltName=issuer:copy\n") - cnf.write("\n") - cnf.write("[ v3_ca_req ]\n") - cnf.write("basicConstraints = critical,CA:true\n") - cnf.write("keyUsage = cRLSign, keyCertSign\n") - cnf.write("nsCertType = sslCA\n") - #cnf.write("subjectAltName=email:copy\n") - cnf.write("\n") - cnf.write("[ v3_cert ]\n") - cnf.write("basicConstraints = CA:false\n") - cnf.write("keyUsage = nonRepudiation, digitalSignature, keyEncipherment\n") - cnf.write("nsCertType = server\n") - cnf.write("subjectKeyIdentifier=hash\n") - cnf.write("authorityKeyIdentifier=keyid:always,issuer\n") - cnf.write("\n") - cnf.write("[ v3_cert_req ]\n") - cnf.write("basicConstraints = CA:false\n") - cnf.write("keyUsage = nonRepudiation, digitalSignature, keyEncipherment\n") - cnf.write("nsCertType = server\n") - cnf.write("\n") - - cnf.close() - - if ca is None: - # Create a new selfsigned certificate + key + return True + + +def dummy_cert(certdir, ca, commonname): + """ + certdir: Certificate directory. + ca: Path to the certificate authority file, or None. + commonname: Common name for the generated certificate. + + Returns True if operation succeeded, False if not. + """ + confpath = os.path.join(certdir, commonname + ".cnf") + reqpath = os.path.join(certdir, commonname + ".req") + certpath = os.path.join(certdir, commonname + ".pem") + + template = open(data.path("resources/cert.cnf")).read() + f = open(confpath, "w").write(template%(dict(commonname=commonname))) + + if ca: + # Create a dummy signed certificate. Uses same key as the signing CA cmd = [ "openssl", "req", "-new", - "-x509", - "-config", path+".cnf", - "-nodes", - "-days", "9999", - "-out", certpath, - "-newkey", "rsa:1024", - "-keyout", certpath, + "-config", confpath, + "-out", reqpath, + "-key", ca, ] - #print " ".join(cmd) - subprocess.call( + ret = subprocess.call( cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, stdin=subprocess.PIPE ) - else: - # Create a dummy signed certificate. Uses same key as the signing CA + if ret: + return False cmd = [ "openssl", - "req", - "-new", - "-config", path+".cnf", - "-out", path+".req", - "-key", ca, + "x509", + "-req", + "-in", reqpath, + "-days", "9999", + "-out", certpath, + "-CA", ca, + "-CAcreateserial", + "-extfile", confpath, + "-extensions", "v3_cert", ] - #print " ".join(cmd) - subprocess.call( + ret = subprocess.call( cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, stdin=subprocess.PIPE ) + if ret: + return False + else: + # Create a new selfsigned certificate + key cmd = [ "openssl", - "x509", - "-req", - "-in", path+".req", + "req", + "-new", + "-x509", + "-config", confpath, + "-nodes", "-days", "9999", "-out", certpath, - "-CA", ca, - "-CAcreateserial", - "-extfile", path+".cnf" + "-newkey", "rsa:1024", + "-keyout", certpath, ] - if newca: - cmd.extend([ - "-extensions", "v3_ca", - ]) - else: - cmd.extend([ - "-extensions", "v3_cert", - ]) - - #print " ".join(cmd) - subprocess.call( + ret = subprocess.call( cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, stdin=subprocess.PIPE ) - + if ret: + return False + return True + def mkdir_p(path): try: @@ -490,11 +441,11 @@ def certificate_option_group(parser): group = optparse.OptionGroup(parser, "SSL") group.add_option( "--cert", action="store", - type = "str", dest="cert", default="~/.mitmproxy/default.pem", + type = "str", dest="cert", default=None, help = "SSL certificate file." ) group.add_option( - "-c", "--cacert", action="store", + "--cacert", action="store", type = "str", dest="cacert", default="~/.mitmproxy/ca.pem", help = "SSL CA certificate file." ) @@ -510,17 +461,17 @@ def certificate_option_group(parser): ) parser.add_option_group(group) -def process_certificate_option_group(options): + +def process_certificate_option_group(parser, options): if options.cert is not None: options.cert = os.path.expanduser(options.cert) if not os.path.exists(options.cert): - print >> sys.stderr, "Creating bogus certificate at %s"%options.cert - utils.make_bogus_cert(options.cert) + parser.error("Manually created certificate does not exist: %s"%options.cert) if options.cacert is not None: options.cacert = os.path.expanduser(options.cacert) if not os.path.exists(options.cacert): - print >> sys.stderr, "Creating bogus CA certificate at %s"%options.cacert - utils.make_bogus_cert(options.cacert, newca=True, commonName="Dummy CA") + print >> sys.stderr, "Creating dummy CA certificate at %s"%options.cacert + dummy_ca(options.cacert) if options.certpath is not None: options.certpath = os.path.expanduser(options.certpath) elif options.cacert is not None: @@ -54,7 +54,7 @@ if __name__ == '__main__': if options.quiet: options.verbose = 0 - utils.process_certificate_option_group(options) + utils.process_certificate_option_group(parser, options) proxy.config = proxy.Config( certfile = options.cert, diff --git a/mitmplayback b/mitmplayback index 6a98d49f..ddfe233b 100755 --- a/mitmplayback +++ b/mitmplayback @@ -56,7 +56,7 @@ if __name__ == '__main__': if options.quiet: options.verbose = 0 - utils.process_certificate_option_group(options) + utils.process_certificate_option_group(parser, options) if options.cache is not None: options.cache = os.path.expanduser(options.cache) @@ -85,7 +85,7 @@ if __name__ == '__main__': options, args = parser.parse_args() - utils.process_certificate_option_group(options) + utils.process_certificate_option_group(parser, options) if options.cache is not None: options.cache = os.path.expanduser(options.cache) @@ -61,7 +61,7 @@ if __name__ == '__main__': if options.quiet: options.verbose = 0 - utils.process_certificate_option_group(options) + utils.process_certificate_option_group(parser, options) proxy.config = proxy.Config( certfile = options.cert, diff --git a/test/test_utils.py b/test/test_utils.py index 5cf81e2e..a52c8e3b 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -217,16 +217,6 @@ class uisSequenceLike(libpry.AutoTree): assert not utils.isSequenceLike(1) -class umake_bogus_cert(libpry.AutoTree): - def test_all(self): - d = self.tmpdir() - path = os.path.join(d, "foo", "cert") - utils.make_bogus_cert(path) - - d = open(path).read() - assert "PRIVATE KEY" in d - assert "CERTIFICATE" in d - class upretty_xmlish(libpry.AutoTree): def test_tagre(self): @@ -284,12 +274,42 @@ class upretty_xmlish(libpry.AutoTree): assert utils.pretty_xmlish(s) == ["gobbledygook"] +class udummy_ca(libpry.AutoTree): + def test_all(self): + d = self.tmpdir() + path = os.path.join(d, "foo/cert.cnf") + assert utils.dummy_ca(path) + assert os.path.exists(path) + + +class udummy_cert(libpry.AutoTree): + def test_with_ca(self): + d = self.tmpdir() + cacert = os.path.join(d, "foo/cert.cnf") + assert utils.dummy_ca(cacert) + assert utils.dummy_cert( + os.path.join(d, "foo"), + cacert, + "foo.com" + ) + assert os.path.exists(os.path.join(d, "foo", "foo.com.pem")) + + def test_no_ca(self): + d = self.tmpdir() + assert utils.dummy_cert( + d, + None, + "foo.com" + ) + assert os.path.exists(os.path.join(d, "foo.com.pem")) + + + tests = [ uformat_timestamp(), - umake_bogus_cert(), uisBin(), uhexdump(), upretty_size(), @@ -299,4 +319,6 @@ tests = [ uHeaders(), uData(), upretty_xmlish(), + udummy_ca(), + udummy_cert(), ] |