Issue #8813: Add SSLContext.verify_flags to change the verification flags

of the context in order to enable certification revocation list (CRL)
checks or strict X509 rules.

4 files changed, 81 insertions, 1 deletions

diff --git a/Lib/ssl.py b/Lib/ssl.py
index 91956c6..7ce097f 100644
--- a/Lib/ssl.py
+++ b/Lib/ssl.py
@@ -102,6 +102,8 @@ from _ssl import (
     SSLSyscallError, SSLEOFError,
     )
 from _ssl import CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
+from _ssl import (VERIFY_DEFAULT, VERIFY_CRL_CHECK_LEAF, VERIFY_CRL_CHECK_CHAIN,
+    VERIFY_X509_STRICT)
 from _ssl import txt2obj as _txt2obj, nid2obj as _nid2obj
 from _ssl import RAND_status, RAND_egd, RAND_add, RAND_bytes, RAND_pseudo_bytes
 
diff --git a/Lib/test/make_ssl_certs.py b/Lib/test/make_ssl_certs.py
index f630813..4251d55 100644
--- a/Lib/test/make_ssl_certs.py
+++ b/Lib/test/make_ssl_certs.py
@@ -28,8 +28,10 @@ req_template = """
     [ CA_default ]
     dir = cadir
     database  = $dir/index.txt
+    crlnumber = $dir/crl.txt
     default_md = sha1
     default_days = 3600
+    default_crl_days = 3600
     certificate = pycacert.pem
     private_key = pycakey.pem
     serial    = $dir/serial
@@ -112,6 +114,8 @@ def make_ca():
     os.mkdir(TMP_CADIR)
     with open(os.path.join('cadir','index.txt'),'a+') as f:
         pass # empty file
+    with open(os.path.join('cadir','crl.txt'),'a+') as f:
+        r.write("00")
     with open(os.path.join('cadir','index.txt.attr'),'w+') as f:
         f.write('unique_subject = no')
 
@@ -129,6 +133,8 @@ def make_ca():
                     '-keyfile', 'pycakey.pem', '-days', '3650',
                     '-selfsign', '-extensions', 'v3_ca', '-infiles', f.name ]
             check_call(['openssl'] + args)
+            args = ['ca', '-config', t.name, '-gencrl', '-out', 'revocation.crl']
+            check_call(['openssl'] + args)
 
 if __name__ == '__main__':
     os.chdir(here)
diff --git a/Lib/test/revocation.crl b/Lib/test/revocation.crl
new file mode 100644
index 0000000..6d89b08
--- /dev/null
+++ b/Lib/test/revocation.crl
@@ -0,0 +1,11 @@
+-----BEGIN X509 CRL-----
+MIIBpjCBjwIBATANBgkqhkiG9w0BAQUFADBNMQswCQYDVQQGEwJYWTEmMCQGA1UE
+CgwdUHl0aG9uIFNvZnR3YXJlIEZvdW5kYXRpb24gQ0ExFjAUBgNVBAMMDW91ci1j
+YS1zZXJ2ZXIXDTEzMTEyMTE3MDg0N1oXDTIzMDkzMDE3MDg0N1qgDjAMMAoGA1Ud
+FAQDAgEAMA0GCSqGSIb3DQEBBQUAA4IBAQCNJXC2mVKauEeN3LlQ3ZtM5gkH3ExH
++i4bmJjtJn497WwvvoIeUdrmVXgJQR93RtV37hZwN0SXMLlNmUZPH4rHhihayw4m
+unCzVj/OhCCY7/TPjKuJ1O/0XhaLBpBVjQN7R/1ujoRKbSia/CD3vcn7Fqxzw7LK
+fSRCKRGTj1CZiuxrphtFchwALXSiFDy9mr2ZKhImcyq1PydfgEzU78APpOkMQsIC
+UNJ/cf3c9emzf+dUtcMEcejQ3mynBo4eIGg1EW42bz4q4hSjzQlKcBV0muw5qXhc
+HOxH2iTFhQ7SrvVuK/dM14rYM4B5mSX3nRC1kNmXpS9j3wJDhuwmjHed
+-----END X509 CRL-----
diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py
index 8016728..2da1386 100644
--- a/Lib/test/test_ssl.py
+++ b/Lib/test/test_ssl.py
@@ -48,6 +48,9 @@ CAFILE_NEURONIO = data_file("capath", "4e1295a3.0")
 CAFILE_CACERT = data_file("capath", "5ed36f99.0")
 
 
+# empty CRL
+CRLFILE = data_file("revocation.crl")
+
 # Two keys and certs signed by the same CA (for SNI tests)
 SIGNED_CERTFILE = data_file("keycert3.pem")
 SIGNED_CERTFILE2 = data_file("keycert4.pem")
@@ -631,7 +634,7 @@ class ContextTests(unittest.TestCase):
             with self.assertRaises(ValueError):
                 ctx.options = 0
 
-    def test_verify(self):
+    def test_verify_mode(self):
         ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
         # Default value
         self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
@@ -646,6 +649,23 @@ class ContextTests(unittest.TestCase):
         with self.assertRaises(ValueError):
             ctx.verify_mode = 42
 
+    def test_verify_flags(self):
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+        # default value by OpenSSL
+        self.assertEqual(ctx.verify_flags, ssl.VERIFY_DEFAULT)
+        ctx.verify_flags = ssl.VERIFY_CRL_CHECK_LEAF
+        self.assertEqual(ctx.verify_flags, ssl.VERIFY_CRL_CHECK_LEAF)
+        ctx.verify_flags = ssl.VERIFY_CRL_CHECK_CHAIN
+        self.assertEqual(ctx.verify_flags, ssl.VERIFY_CRL_CHECK_CHAIN)
+        ctx.verify_flags = ssl.VERIFY_DEFAULT
+        self.assertEqual(ctx.verify_flags, ssl.VERIFY_DEFAULT)
+        # supports any value
+        ctx.verify_flags = ssl.VERIFY_CRL_CHECK_LEAF | ssl.VERIFY_X509_STRICT
+        self.assertEqual(ctx.verify_flags,
+                         ssl.VERIFY_CRL_CHECK_LEAF | ssl.VERIFY_X509_STRICT)
+        with self.assertRaises(TypeError):
+            ctx.verify_flags = None
+
     def test_load_cert_chain(self):
         ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
         # Combined key and cert in a single file
@@ -1771,6 +1791,47 @@ else:
                 self.assertLess(before, after)
                 s.close()
 
+        def test_crl_check(self):
+            if support.verbose:
+                sys.stdout.write("\n")
+
+            server_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+            server_context.load_cert_chain(SIGNED_CERTFILE)
+
+            context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
+            context.verify_mode = ssl.CERT_REQUIRED
+            context.load_verify_locations(SIGNING_CA)
+            context.verify_mode = ssl.CERT_REQUIRED
+            context.verify_flags = ssl.VERIFY_DEFAULT
+
+            # VERIFY_DEFAULT should pass
+            server = ThreadedEchoServer(context=server_context, chatty=True)
+            with server:
+                with context.wrap_socket(socket.socket()) as s:
+                    s.connect((HOST, server.port))
+                    cert = s.getpeercert()
+                    self.assertTrue(cert, "Can't get peer certificate.")
+
+            # VERIFY_CRL_CHECK_LEAF without a loaded CRL file fails
+            context.verify_flags = ssl.VERIFY_CRL_CHECK_LEAF
+
+            server = ThreadedEchoServer(context=server_context, chatty=True)
+            with server:
+                with context.wrap_socket(socket.socket()) as s:
+                    with self.assertRaisesRegex(ssl.SSLError,
+                                                "certificate verify failed"):
+                        s.connect((HOST, server.port))
+
+            # now load a CRL file. The CRL file is signed by the CA.
+            context.load_verify_locations(CRLFILE)
+
+            server = ThreadedEchoServer(context=server_context, chatty=True)
+            with server:
+                with context.wrap_socket(socket.socket()) as s:
+                    s.connect((HOST, server.port))
+                    cert = s.getpeercert()
+                    self.assertTrue(cert, "Can't get peer certificate.")
+
         def test_empty_cert(self):
             """Connecting with an empty cert file"""
             bad_cert_test(os.path.join(os.path.dirname(__file__) or os.curdir,