diff options
| -rw-r--r-- | netlib/tcp.py | 23 | ||||
| -rw-r--r-- | test/data/not-server.crt | 15 | ||||
| -rw-r--r-- | test/test_tcp.py | 53 |
3 files changed, 91 insertions, 0 deletions
diff --git a/netlib/tcp.py b/netlib/tcp.py index 897e3e65..953cef6e 100644 --- a/netlib/tcp.py +++ b/netlib/tcp.py @@ -24,6 +24,7 @@ TLSv1_2_METHOD = SSL.TLSv1_2_METHOD OP_NO_SSLv2 = SSL.OP_NO_SSLv2 OP_NO_SSLv3 = SSL.OP_NO_SSLv3 +VERIFY_NONE = SSL.VERIFY_NONE class NetLibError(Exception): @@ -374,6 +375,9 @@ class _Connection(object): def _create_ssl_context(self, method=SSLv23_METHOD, options=(OP_NO_SSLv2 | OP_NO_SSLv3), + verify_options=VERIFY_NONE, + ca_path=None, + ca_pemfile=None, cipher_list=None, alpn_protos=None, alpn_select=None, @@ -381,6 +385,9 @@ class _Connection(object): """ :param method: One of SSLv2_METHOD, SSLv3_METHOD, SSLv23_METHOD, TLSv1_METHOD, TLSv1_1_METHOD, or TLSv1_2_METHOD :param options: A bit field consisting of OpenSSL.SSL.OP_* values + :param verify_options: A bit field consisting of OpenSSL.SSL.VERIFY_* values + :param ca_path: Path to a directory of trusted CA certificates prepared using the c_rehash tool + :param ca_pemfile: Path to a PEM formatted trusted CA certificate :param cipher_list: A textual OpenSSL cipher list, see https://www.openssl.org/docs/apps/ciphers.html :rtype : SSL.Context """ @@ -389,6 +396,19 @@ class _Connection(object): if options is not None: context.set_options(options) + # Verify Options (NONE/PEER/PEER|FAIL_IF_... and trusted CAs) + if verify_options is not None and verify_options is not VERIFY_NONE: + def verify_cert(conn, cert, errno, err_depth, is_cert_verified): + if is_cert_verified: + return True + raise NetLibError( + "Upstream certificate validation failed at depth: %s with error number: %s" % + (err_depth, errno)) + + context.set_verify(verify_options, verify_cert) + if ca_path is not None or ca_pemfile is not None: + context.load_verify_locations(ca_pemfile, ca_path) + # Workaround for # https://github.com/pyca/pyopenssl/issues/190 # https://github.com/mitmproxy/mitmproxy/issues/472 @@ -462,6 +482,9 @@ class TCPClient(_Connection): cert: Path to a file containing both client cert and private key. options: A bit field consisting of OpenSSL.SSL.OP_* values + verify_options: A bit field consisting of OpenSSL.SSL.VERIFY_* values + ca_path: Path to a directory of trusted CA certificates prepared using the c_rehash tool + ca_pemfile: Path to a PEM formatted trusted CA certificate """ context = self.create_ssl_context( alpn_protos=alpn_protos, diff --git a/test/data/not-server.crt b/test/data/not-server.crt new file mode 100644 index 00000000..08c015c2 --- /dev/null +++ b/test/data/not-server.crt @@ -0,0 +1,15 @@ +-----BEGIN CERTIFICATE----- +MIICRTCCAa4CCQD/j4qq1h3iCjANBgkqhkiG9w0BAQUFADBnMQswCQYDVQQGEwJV +UzELMAkGA1UECBMCQ0ExETAPBgNVBAcTCFNvbWVDaXR5MRcwFQYDVQQKEw5Ob3RU +aGVSaWdodE9yZzELMAkGA1UECxMCTkExEjAQBgNVBAMTCU5vdFNlcnZlcjAeFw0x +NTA2MTMwMTE2MDZaFw0yNTA2MTAwMTE2MDZaMGcxCzAJBgNVBAYTAlVTMQswCQYD +VQQIEwJDQTERMA8GA1UEBxMIU29tZUNpdHkxFzAVBgNVBAoTDk5vdFRoZVJpZ2h0 +T3JnMQswCQYDVQQLEwJOQTESMBAGA1UEAxMJTm90U2VydmVyMIGfMA0GCSqGSIb3 +DQEBAQUAA4GNADCBiQKBgQDPkJlXAOCMKF0R7aDn5QJ7HtrJgOUDk/LpbhKhRZZR +dRGnJ4/HQxYYHh9k/4yZamYcvQPUxvFJt7UJUocf+84LUcIusUk7GvJMgsMVtFMq +7UKNXBN5tl3oOtoFDWGMZ8ksaIxS6oW3V/9v2WgU23PfvwE0EZqy+QhMLZZP5GOH +RwIDAQABMA0GCSqGSIb3DQEBBQUAA4GBAJI6UtMKdCS2ghjqhAek2W1rt9u+Wuvx +776WYm5VyrJEtBDc/axLh0OteXzy/A31JrYe15fnVWIeFbDF0Ief9/Ezv6Jn+Pk8 +DErw5IHk2B399O4K3L3Eig06piu7uf3vE4l8ZanY02ZEnw7DyL6kmG9lX98VGenF +uXPfu3yxKbR4 +-----END CERTIFICATE----- diff --git a/test/test_tcp.py b/test/test_tcp.py index 8aa34d2b..0cecaaa2 100644 --- a/test/test_tcp.py +++ b/test/test_tcp.py @@ -171,6 +171,59 @@ class TestSSLv3Only(test.ServerTestBase): tutils.raises(tcp.NetLibError, c.convert_to_ssl, sni="foo.com") +class TestSSLUpstreamCertVerification(test.ServerTestBase): + handler = EchoHandler + + ssl = dict( + cert=tutils.test_data.path("data/server.crt") + ) + + def test_mode_default(self): + c = tcp.TCPClient(("127.0.0.1", self.port)) + c.connect() + + c.convert_to_ssl() + + testval = "echo!\n" + c.wfile.write(testval) + c.wfile.flush() + assert c.rfile.readline() == testval + + def test_mode_none(self): + c = tcp.TCPClient(("127.0.0.1", self.port)) + c.connect() + + c.convert_to_ssl(verify_options=SSL.VERIFY_NONE) + + testval = "echo!\n" + c.wfile.write(testval) + c.wfile.flush() + assert c.rfile.readline() == testval + + def test_mode_strict_w_bad_cert(self): + c = tcp.TCPClient(("127.0.0.1", self.port)) + c.connect() + + tutils.raises( + tcp.NetLibError, + c.convert_to_ssl, + verify_options=SSL.VERIFY_PEER | SSL.VERIFY_FAIL_IF_NO_PEER_CERT, + ca_pemfile=tutils.test_data.path("data/not-server.crt")) + + def test_mode_strict_w_cert(self): + c = tcp.TCPClient(("127.0.0.1", self.port)) + c.connect() + + c.convert_to_ssl( + verify_options=SSL.VERIFY_PEER | SSL.VERIFY_FAIL_IF_NO_PEER_CERT, + ca_pemfile=tutils.test_data.path("data/server.crt")) + + testval = "echo!\n" + c.wfile.write(testval) + c.wfile.flush() + assert c.rfile.readline() == testval + + class TestSSLClientCert(test.ServerTestBase): class handler(tcp.BaseHandler): |