Adding support for upstream certificate validation when using SSL/TLS with an

instance of TCPClient.

netlib/tcp.py

@@ -21,6 +21,7 @@ SSLv23_METHOD = SSL.SSLv23_METHOD
 TLSv1_METHOD = SSL.TLSv1_METHOD
 OP_NO_SSLv2 = SSL.OP_NO_SSLv2
 OP_NO_SSLv3 = SSL.OP_NO_SSLv3
+VERIFY_NONE = SSL.VERIFY_NONE
 
 
 class NetLibError(Exception):
@@ -371,6 +372,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,
@@ -378,6 +382,9 @@ class _Connection(object):
         """
         :param method: One of SSLv2_METHOD, SSLv3_METHOD, SSLv23_METHOD, TLSv1_METHOD or TLSv1_1_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
         """
@@ -386,6 +393,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
@@ -458,6 +478,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,

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-----

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):