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

improve invalid certificate ux

Browse Source
This commit is contained in:
Maximilian Hils 2016-07-27 21:01:28 -07:00
parent 17fdb841f0
commit 8b325fd65a
9 changed files with 70 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
mitmproxy/console/master.py
View File

@ -277,11 +277,11 @@ class ConsoleMaster(flow.FlowMaster):
if self.options.verbosity < utils.log_tier(level):
return
if level == "error":
if level in ("error", "warn"):
signals.status_message.send(
message = "Error: %s" % str(e)
message = "{}: {}".format(level.title(), e)
)
e = urwid.Text(("error", str(e)))
e = urwid.Text((level, str(e)))
else:
e = urwid.Text(str(e))
self.logbuffer.append(e)

1
mitmproxy/console/options.py
View File

@ -8,7 +8,6 @@ from mitmproxy.console import grideditor
from mitmproxy.console import palettes
from mitmproxy.console import select
from mitmproxy.console import signals
from OpenSSL import SSL
footer = [
('heading_key', "enter/space"), ":toggle ",

2
mitmproxy/dump.py
View File

@ -104,7 +104,7 @@ class DumpMaster(flow.FlowMaster):
click.secho(
e,
file=self.options.tfile,
fg="red" if level == "error" else None,
fg=dict(error="red", warn="yellow").get(level),
dim=(level == "debug"),
err=(level == "error")
)

6
mitmproxy/exceptions.py
View File

@ -44,6 +44,12 @@ class ClientHandshakeException(TlsProtocolException):
self.server = server
class InvalidServerCertificate(TlsProtocolException):
def __repr__(self):
# In contrast to most others, this is a user-facing error which needs to look good.
return str(self)
class Socks5ProtocolException(ProtocolException):
pass

2
mitmproxy/models/http.py
View File

@ -225,7 +225,7 @@ class HTTPFlow(Flow):
def make_error_response(status_code, message, headers=None):
response = status_codes.RESPONSES.get(status_code, "Unknown").encode()
response = status_codes.RESPONSES.get(status_code, "Unknown")
body = """
<html>
<head>

21
mitmproxy/protocol/tls.py
View File

@ -543,25 +543,12 @@ class TlsLayer(base.Layer):
)
tls_cert_err = self.server_conn.ssl_verification_error
if tls_cert_err is not None:
self.log(
"TLS verification failed for upstream server at depth %s with error: %s" %
(tls_cert_err['depth'], tls_cert_err['errno']),
"error")
self.log("Ignoring server verification error, continuing with connection", "error")
self.log(str(tls_cert_err), "warn")
self.log("Ignoring server verification error, continuing with connection", "warn")
except netlib.exceptions.InvalidCertificateException as e:
tls_cert_err = self.server_conn.ssl_verification_error
self.log(
"TLS verification failed for upstream server at depth %s with error: %s" %
(tls_cert_err['depth'], tls_cert_err['errno']),
"error")
self.log("Aborting connection attempt", "error")
six.reraise(
exceptions.TlsProtocolException,
exceptions.TlsProtocolException("Cannot establish TLS with {address} (sni: {sni}): {e}".format(
address=repr(self.server_conn.address),
sni=self.server_sni,
e=repr(e),
)),
exceptions.InvalidServerCertificate,
exceptions.InvalidServerCertificate(str(e)),
sys.exc_info()[2]
)
except netlib.exceptions.TlsException as e:

7
mitmproxy/proxy/server.py
View File

@ -125,11 +125,14 @@ class ConnectionHandler(object):
self.log(
"Client Handshake failed. "
"The client may not trust the proxy's certificate for {}.".format(e.server),
"error"
"warn"
)
self.log(repr(e), "debug")
elif isinstance(e, exceptions.InvalidServerCertificate):
self.log(str(e), "warn")
self.log("Invalid certificate, closing connection. Pass --insecure to disable validation.", "warn")
else:
self.log(repr(e), "info")
self.log(repr(e), "warn")
self.log(traceback.format_exc(), "debug")
# If an error propagates to the topmost level,

61
netlib/tcp.py
View File

@ -8,6 +8,10 @@ import time
import traceback
import binascii
from typing import Optional # noqa
from netlib import strutils
from six.moves import range
import certifi
@ -35,7 +39,7 @@ EINTR = 4
if os.environ.get("NO_ALPN"):
HAS_ALPN = False
else:
HAS_ALPN = OpenSSL._util.lib.Cryptography_HAS_ALPN
HAS_ALPN = SSL._lib.Cryptography_HAS_ALPN
# To enable all SSL methods use: SSLv23
# then add options to disable certain methods
@ -287,16 +291,7 @@ class Reader(_FileLike):
raise exceptions.TcpException(repr(e))
elif isinstance(self.o, SSL.Connection):
try:
if tuple(int(x) for x in OpenSSL.__version__.split(".")[:2]) > (0, 15):
return self.o.recv(length, socket.MSG_PEEK)
else:
# TODO: remove once a new version is released
# Polyfill for pyOpenSSL <= 0.15.1
# Taken from https://github.com/pyca/pyopenssl/commit/1d95dea7fea03c7c0df345a5ea30c12d8a0378d2
buf = SSL._ffi.new("char[]", length)
result = SSL._lib.SSL_peek(self.o._ssl, buf, length)
self.o._raise_ssl_error(self.o._ssl, result)
return SSL._ffi.buffer(buf, result)[:]
except SSL.Error as e:
six.reraise(exceptions.TlsException, exceptions.TlsException(str(e)), sys.exc_info()[2])
else:
@ -436,6 +431,23 @@ def close_socket(sock):
sock.close()
class SSLVerificationError:
def __init__(self, errno, depth, message=None):
self.errno = errno
self.depth = depth
if message:
self.message = message
else:
self.message = strutils.native(SSL._ffi.string(SSL._lib.X509_verify_cert_error_string(errno)), "utf8")
def __str__(self):
return "Certificate Verification Error: {} (errno: {}, depth: {})".format(
self.message,
self.errno,
self.depth
)
class _Connection(object):
rbufsize = -1
@ -511,6 +523,7 @@ class _Connection(object):
alpn_protos=None,
alpn_select=None,
alpn_select_callback=None,
sni=None,
):
"""
Creates an SSL Context.
@ -532,8 +545,14 @@ class _Connection(object):
if verify_options is not None:
def verify_cert(conn, x509, errno, err_depth, is_cert_verified):
if not is_cert_verified:
self.ssl_verification_error = dict(errno=errno,
depth=err_depth)
self.ssl_verification_error = exceptions.InvalidCertificateException(
"Certificate Verification Error for {}: {} (errno: {}, depth: {})".format(
sni,
strutils.native(SSL._ffi.string(SSL._lib.X509_verify_cert_error_string(errno)), "utf8"),
errno,
err_depth
)
)
return is_cert_verified
context.set_verify(verify_options, verify_cert)
@ -609,7 +628,7 @@ class TCPClient(_Connection):
self.source_address = source_address
self.cert = None
self.server_certs = []
self.ssl_verification_error = None
self.ssl_verification_error = None # type: Optional[SSLVerificationError]
self.sni = None
@property
@ -671,6 +690,7 @@ class TCPClient(_Connection):
context = self.create_ssl_context(
alpn_protos=alpn_protos,
sni=sni,
**sslctx_kwargs
)
self.connection = SSL.Connection(context, self.connection)
@ -682,14 +702,14 @@ class TCPClient(_Connection):
self.connection.do_handshake()
except SSL.Error as v:
if self.ssl_verification_error:
raise exceptions.InvalidCertificateException("SSL handshake error: %s" % repr(v))
raise self.ssl_verification_error
else:
raise exceptions.TlsException("SSL handshake error: %s" % repr(v))
else:
# Fix for pre v1.0 OpenSSL, which doesn't throw an exception on
# certificate validation failure
if verification_mode == SSL.VERIFY_PEER and self.ssl_verification_error is not None:
raise exceptions.InvalidCertificateException("SSL handshake error: certificate verify failed")
if verification_mode == SSL.VERIFY_PEER and self.ssl_verification_error:
raise self.ssl_verification_error
self.cert = certutils.SSLCert(self.connection.get_peer_certificate())
@ -710,9 +730,14 @@ class TCPClient(_Connection):
hostname = "no-hostname"
ssl_match_hostname.match_hostname(crt, hostname)
except (ValueError, ssl_match_hostname.CertificateError) as e:
self.ssl_verification_error = dict(depth=0, errno="Invalid Hostname")
self.ssl_verification_error = exceptions.InvalidCertificateException(
"Certificate Verification Error for {}: {}".format(
sni or repr(self.address),
str(e)
)
)
if verification_mode == SSL.VERIFY_PEER:
raise exceptions.InvalidCertificateException("Presented certificate for {} is not valid: {}".format(sni, str(e)))
raise self.ssl_verification_error
self.ssl_established = True
self.rfile.set_descriptor(self.connection)

12
test/netlib/test_tcp.py
View File

@ -213,7 +213,7 @@ class TestSSLUpstreamCertVerificationWBadServerCert(tservers.ServerTestBase):
# Verification errors should be saved even if connection isn't aborted
# aborted
assert c.ssl_verification_error is not None
assert c.ssl_verification_error
testval = b"echo!\n"
c.wfile.write(testval)
@ -226,7 +226,7 @@ class TestSSLUpstreamCertVerificationWBadServerCert(tservers.ServerTestBase):
c.convert_to_ssl(verify_options=SSL.VERIFY_NONE)
# Verification errors should be saved even if connection isn't aborted
assert c.ssl_verification_error is not None
assert c.ssl_verification_error
testval = b"echo!\n"
c.wfile.write(testval)
@ -243,11 +243,11 @@ class TestSSLUpstreamCertVerificationWBadServerCert(tservers.ServerTestBase):
ca_pemfile=tutils.test_data.path("data/verificationcerts/trusted-root.crt")
)
assert c.ssl_verification_error is not None
assert c.ssl_verification_error
# Unknown issuing certificate authority for first certificate
assert c.ssl_verification_error['errno'] == 18
assert c.ssl_verification_error['depth'] == 0
assert "errno: 18" in str(c.ssl_verification_error)
assert "depth: 0" in str(c.ssl_verification_error)
class TestSSLUpstreamCertVerificationWBadHostname(tservers.ServerTestBase):
@ -276,7 +276,7 @@ class TestSSLUpstreamCertVerificationWBadHostname(tservers.ServerTestBase):
verify_options=SSL.VERIFY_PEER,
ca_pemfile=tutils.test_data.path("data/verificationcerts/trusted-root.crt")
)
assert c.ssl_verification_error is not None
assert c.ssl_verification_error
class TestSSLUpstreamCertVerificationWValidCertChain(tservers.ServerTestBase):