diff options
Diffstat (limited to 'lib-python/3/test/test_ssl.py')
| -rw-r--r-- | lib-python/3/test/test_ssl.py | 224 |
1 files changed, 160 insertions, 64 deletions
diff --git a/lib-python/3/test/test_ssl.py b/lib-python/3/test/test_ssl.py index 1556a4dcd7..47b11603ce 100644 --- a/lib-python/3/test/test_ssl.py +++ b/lib-python/3/test/test_ssl.py @@ -19,6 +19,7 @@ import weakref import platform import functools import sysconfig +import functools try: import ctypes except ImportError: @@ -142,6 +143,87 @@ OP_CIPHER_SERVER_PREFERENCE = getattr(ssl, "OP_CIPHER_SERVER_PREFERENCE", 0) OP_ENABLE_MIDDLEBOX_COMPAT = getattr(ssl, "OP_ENABLE_MIDDLEBOX_COMPAT", 0) +def has_tls_protocol(protocol): + """Check if a TLS protocol is available and enabled + + :param protocol: enum ssl._SSLMethod member or name + :return: bool + """ + if isinstance(protocol, str): + assert protocol.startswith('PROTOCOL_') + protocol = getattr(ssl, protocol, None) + if protocol is None: + return False + if protocol in { + ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS_SERVER, + ssl.PROTOCOL_TLS_CLIENT + }: + # auto-negotiate protocols are always available + return True + name = protocol.name + return has_tls_version(name[len('PROTOCOL_'):]) + + +@functools.lru_cache() +def has_tls_version(version): + """Check if a TLS/SSL version is enabled + + :param version: TLS version name or ssl.TLSVersion member + :return: bool + """ + if version == "SSLv2": + # never supported and not even in TLSVersion enum + return False + + if isinstance(version, str): + version = ssl.TLSVersion.__members__[version] + + # check compile time flags like ssl.HAS_TLSv1_2 + if not getattr(ssl, f'HAS_{version.name}'): + return False + + # check runtime and dynamic crypto policy settings. A TLS version may + # be compiled in but disabled by a policy or config option. + ctx = ssl.SSLContext() + if ( + hasattr(ctx, 'minimum_version') and + ctx.minimum_version != ssl.TLSVersion.MINIMUM_SUPPORTED and + version < ctx.minimum_version + ): + return False + if ( + hasattr(ctx, 'maximum_version') and + ctx.maximum_version != ssl.TLSVersion.MAXIMUM_SUPPORTED and + version > ctx.maximum_version + ): + return False + + return True + + +def requires_tls_version(version): + """Decorator to skip tests when a required TLS version is not available + + :param version: TLS version name or ssl.TLSVersion member + :return: + """ + def decorator(func): + @functools.wraps(func) + def wrapper(*args, **kw): + if not has_tls_version(version): + raise unittest.SkipTest(f"{version} is not available.") + else: + return func(*args, **kw) + return wrapper + return decorator + + +requires_minimum_version = unittest.skipUnless( + hasattr(ssl.SSLContext, 'minimum_version'), + "required OpenSSL >= 1.1.0g" +) + + def handle_error(prefix): exc_format = ' '.join(traceback.format_exception(*sys.exc_info())) if support.verbose: @@ -418,7 +500,7 @@ class BasicSocketTests(unittest.TestCase): ('email', 'null@python.org\x00user@example.org'), ('URI', 'http://null.python.org\x00http://example.org'), ('IP Address', '192.0.2.1'), - ('IP Address', '2001:DB8:0:0:0:0:0:1\n')) + ('IP Address', '2001:DB8:0:0:0:0:0:1')) else: # OpenSSL 0.9.7 doesn't support IPv6 addresses in subjectAltName san = (('DNS', 'altnull.python.org\x00example.com'), @@ -445,7 +527,7 @@ class BasicSocketTests(unittest.TestCase): (('commonName', 'dirname example'),))), ('URI', 'https://www.python.org/'), ('IP Address', '127.0.0.1'), - ('IP Address', '0:0:0:0:0:0:0:1\n'), + ('IP Address', '0:0:0:0:0:0:0:1'), ('Registered ID', '1.2.3.4.5') ) ) @@ -472,11 +554,11 @@ class BasicSocketTests(unittest.TestCase): # Some sanity checks follow # >= 0.9 self.assertGreaterEqual(n, 0x900000) - # < 3.0 - self.assertLess(n, 0x30000000) + # < 4.0 + self.assertLess(n, 0x40000000) major, minor, fix, patch, status = t - self.assertGreaterEqual(major, 0) - self.assertLess(major, 3) + self.assertGreaterEqual(major, 1) + self.assertLess(major, 4) self.assertGreaterEqual(minor, 0) self.assertLess(minor, 256) self.assertGreaterEqual(fix, 0) @@ -1124,22 +1206,32 @@ class ContextTests(unittest.TestCase): with self.assertRaises(AttributeError): ctx.hostname_checks_common_name = True - @unittest.skipUnless(hasattr(ssl.SSLContext, 'minimum_version'), - "required OpenSSL 1.1.0g") + @requires_minimum_version + @unittest.skipIf(IS_LIBRESSL, "see bpo-34001") def test_min_max_version(self): ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) # OpenSSL default is MINIMUM_SUPPORTED, however some vendors like # Fedora override the setting to TLS 1.0. + minimum_range = { + # stock OpenSSL + ssl.TLSVersion.MINIMUM_SUPPORTED, + # Fedora 29 uses TLS 1.0 by default + ssl.TLSVersion.TLSv1, + # RHEL 8 uses TLS 1.2 by default + ssl.TLSVersion.TLSv1_2 + } + maximum_range = { + # stock OpenSSL + ssl.TLSVersion.MAXIMUM_SUPPORTED, + # Fedora 32 uses TLS 1.3 by default + ssl.TLSVersion.TLSv1_3 + } + self.assertIn( - ctx.minimum_version, - {ssl.TLSVersion.MINIMUM_SUPPORTED, - # Fedora 29 uses TLS 1.0 by default - ssl.TLSVersion.TLSv1, - # RHEL 8 uses TLS 1.2 by default - ssl.TLSVersion.TLSv1_2} + ctx.minimum_version, minimum_range ) - self.assertEqual( - ctx.maximum_version, ssl.TLSVersion.MAXIMUM_SUPPORTED + self.assertIn( + ctx.maximum_version, maximum_range ) ctx.minimum_version = ssl.TLSVersion.TLSv1_1 @@ -1182,8 +1274,8 @@ class ContextTests(unittest.TestCase): ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_1) - self.assertEqual( - ctx.minimum_version, ssl.TLSVersion.MINIMUM_SUPPORTED + self.assertIn( + ctx.minimum_version, minimum_range ) self.assertEqual( ctx.maximum_version, ssl.TLSVersion.MAXIMUM_SUPPORTED @@ -2724,6 +2816,8 @@ class ThreadedTests(unittest.TestCase): for protocol in PROTOCOLS: if protocol in {ssl.PROTOCOL_TLS_CLIENT, ssl.PROTOCOL_TLS_SERVER}: continue + if not has_tls_protocol(protocol): + continue with self.subTest(protocol=ssl._PROTOCOL_NAMES[protocol]): context = ssl.SSLContext(protocol) context.load_cert_chain(CERTFILE) @@ -3015,7 +3109,7 @@ class ThreadedTests(unittest.TestCase): else: self.fail("Use of invalid cert should have failed!") - @unittest.skipUnless(ssl.HAS_TLSv1_3, "Test needs TLS 1.3") + @requires_tls_version('TLSv1_3') def test_wrong_cert_tls13(self): client_context, server_context, hostname = testing_context() # load client cert that is not signed by trusted CA @@ -3110,9 +3204,7 @@ class ThreadedTests(unittest.TestCase): self.assertIn(msg, repr(e)) self.assertIn('certificate verify failed', repr(e)) - @skip_if_broken_ubuntu_ssl - @unittest.skipUnless(hasattr(ssl, 'PROTOCOL_SSLv2'), - "OpenSSL is compiled without SSLv2 support") + @requires_tls_version('SSLv2') def test_protocol_sslv2(self): """Connecting to an SSLv2 server with various client options""" if support.verbose: @@ -3121,7 +3213,7 @@ class ThreadedTests(unittest.TestCase): try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv2, True, ssl.CERT_REQUIRED) try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_TLS, False) - if hasattr(ssl, 'PROTOCOL_SSLv3'): + if has_tls_version('SSLv3'): try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_SSLv3, False) try_protocol_combo(ssl.PROTOCOL_SSLv2, ssl.PROTOCOL_TLSv1, False) # SSLv23 client with specific SSL options @@ -3139,7 +3231,7 @@ class ThreadedTests(unittest.TestCase): """Connecting to an SSLv23 server with various client options""" if support.verbose: sys.stdout.write("\n") - if hasattr(ssl, 'PROTOCOL_SSLv2'): + if has_tls_version('SSLv2'): try: try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv2, True) except OSError as x: @@ -3148,35 +3240,36 @@ class ThreadedTests(unittest.TestCase): sys.stdout.write( " SSL2 client to SSL23 server test unexpectedly failed:\n %s\n" % str(x)) - if hasattr(ssl, 'PROTOCOL_SSLv3'): + if has_tls_version('SSLv3'): try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv3, False) try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS, True) - try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, 'TLSv1') + if has_tls_version('TLSv1'): + try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, 'TLSv1') - if hasattr(ssl, 'PROTOCOL_SSLv3'): + if has_tls_version('SSLv3'): try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv3, False, ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS, True, ssl.CERT_OPTIONAL) - try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_OPTIONAL) + if has_tls_version('TLSv1'): + try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_OPTIONAL) - if hasattr(ssl, 'PROTOCOL_SSLv3'): + if has_tls_version('SSLv3'): try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv3, False, ssl.CERT_REQUIRED) try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS, True, ssl.CERT_REQUIRED) - try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_REQUIRED) + if has_tls_version('TLSv1'): + try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_REQUIRED) # Server with specific SSL options - if hasattr(ssl, 'PROTOCOL_SSLv3'): + if has_tls_version('SSLv3'): try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv3, False, server_options=ssl.OP_NO_SSLv3) # Will choose TLSv1 try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS, True, server_options=ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3) - try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, False, - server_options=ssl.OP_NO_TLSv1) - + if has_tls_version('TLSv1'): + try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, False, + server_options=ssl.OP_NO_TLSv1) - @skip_if_broken_ubuntu_ssl - @unittest.skipUnless(hasattr(ssl, 'PROTOCOL_SSLv3'), - "OpenSSL is compiled without SSLv3 support") + @requires_tls_version('SSLv3') def test_protocol_sslv3(self): """Connecting to an SSLv3 server with various client options""" if support.verbose: @@ -3184,7 +3277,7 @@ class ThreadedTests(unittest.TestCase): try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, 'SSLv3') try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, 'SSLv3', ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, 'SSLv3', ssl.CERT_REQUIRED) - if hasattr(ssl, 'PROTOCOL_SSLv2'): + if has_tls_version('SSLv2'): try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv2, False) try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_TLS, False, client_options=ssl.OP_NO_SSLv3) @@ -3194,7 +3287,7 @@ class ThreadedTests(unittest.TestCase): try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_TLS, False, client_options=ssl.OP_NO_SSLv2) - @skip_if_broken_ubuntu_ssl + @requires_tls_version('TLSv1') def test_protocol_tlsv1(self): """Connecting to a TLSv1 server with various client options""" if support.verbose: @@ -3202,36 +3295,32 @@ class ThreadedTests(unittest.TestCase): try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, 'TLSv1') try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_OPTIONAL) try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_REQUIRED) - if hasattr(ssl, 'PROTOCOL_SSLv2'): + if has_tls_version('SSLv2'): try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv2, False) - if hasattr(ssl, 'PROTOCOL_SSLv3'): + if has_tls_version('SSLv3'): try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv3, False) try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLS, False, client_options=ssl.OP_NO_TLSv1) - @skip_if_broken_ubuntu_ssl - @unittest.skipUnless(hasattr(ssl, "PROTOCOL_TLSv1_1"), - "TLS version 1.1 not supported.") + @requires_tls_version('TLSv1_1') def test_protocol_tlsv1_1(self): """Connecting to a TLSv1.1 server with various client options. Testing against older TLS versions.""" if support.verbose: sys.stdout.write("\n") try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLSv1_1, 'TLSv1.1') - if hasattr(ssl, 'PROTOCOL_SSLv2'): + if has_tls_version('SSLv2'): try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_SSLv2, False) - if hasattr(ssl, 'PROTOCOL_SSLv3'): + if has_tls_version('SSLv3'): try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_SSLv3, False) try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLS, False, client_options=ssl.OP_NO_TLSv1_1) try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1_1, 'TLSv1.1') - try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLSv1, False) - try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1_1, False) + try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLSv1_2, False) + try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_TLSv1_1, False) - @skip_if_broken_ubuntu_ssl - @unittest.skipUnless(hasattr(ssl, "PROTOCOL_TLSv1_2"), - "TLS version 1.2 not supported.") + @requires_tls_version('TLSv1_2') def test_protocol_tlsv1_2(self): """Connecting to a TLSv1.2 server with various client options. Testing against older TLS versions.""" @@ -3240,9 +3329,9 @@ class ThreadedTests(unittest.TestCase): try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_TLSv1_2, 'TLSv1.2', server_options=ssl.OP_NO_SSLv3|ssl.OP_NO_SSLv2, client_options=ssl.OP_NO_SSLv3|ssl.OP_NO_SSLv2,) - if hasattr(ssl, 'PROTOCOL_SSLv2'): + if has_tls_version('SSLv2'): try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_SSLv2, False) - if hasattr(ssl, 'PROTOCOL_SSLv3'): + if has_tls_version('SSLv3'): try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_SSLv3, False) try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_TLS, False, client_options=ssl.OP_NO_TLSv1_2) @@ -3689,7 +3778,7 @@ class ThreadedTests(unittest.TestCase): self.assertIs(s.version(), None) self.assertIs(s._sslobj, None) s.connect((HOST, server.port)) - if IS_OPENSSL_1_1_1 and ssl.HAS_TLSv1_3: + if IS_OPENSSL_1_1_1 and has_tls_version('TLSv1_3'): self.assertEqual(s.version(), 'TLSv1.3') elif ssl.OPENSSL_VERSION_INFO >= (1, 0, 2): self.assertEqual(s.version(), 'TLSv1.2') @@ -3698,8 +3787,7 @@ class ThreadedTests(unittest.TestCase): self.assertIs(s._sslobj, None) self.assertIs(s.version(), None) - @unittest.skipUnless(ssl.HAS_TLSv1_3, - "test requires TLSv1.3 enabled OpenSSL") + @requires_tls_version('TLSv1_3') def test_tls1_3(self): context = ssl.SSLContext(ssl.PROTOCOL_TLS) context.load_cert_chain(CERTFILE) @@ -3716,9 +3804,9 @@ class ThreadedTests(unittest.TestCase): }) self.assertEqual(s.version(), 'TLSv1.3') - @unittest.skipUnless(hasattr(ssl.SSLContext, 'minimum_version'), - "required OpenSSL 1.1.0g") - def test_min_max_version(self): + @requires_minimum_version + @requires_tls_version('TLSv1_2') + def test_min_max_version_tlsv1_2(self): client_context, server_context, hostname = testing_context() # client TLSv1.0 to 1.2 client_context.minimum_version = ssl.TLSVersion.TLSv1 @@ -3733,7 +3821,13 @@ class ThreadedTests(unittest.TestCase): s.connect((HOST, server.port)) self.assertEqual(s.version(), 'TLSv1.2') + @requires_minimum_version + @requires_tls_version('TLSv1_1') + def test_min_max_version_tlsv1_1(self): + client_context, server_context, hostname = testing_context() # client 1.0 to 1.2, server 1.0 to 1.1 + client_context.minimum_version = ssl.TLSVersion.TLSv1 + client_context.maximum_version = ssl.TLSVersion.TLSv1_2 server_context.minimum_version = ssl.TLSVersion.TLSv1 server_context.maximum_version = ssl.TLSVersion.TLSv1_1 @@ -3743,6 +3837,10 @@ class ThreadedTests(unittest.TestCase): s.connect((HOST, server.port)) self.assertEqual(s.version(), 'TLSv1.1') + @requires_minimum_version + @requires_tls_version('TLSv1_2') + def test_min_max_version_mismatch(self): + client_context, server_context, hostname = testing_context() # client 1.0, server 1.2 (mismatch) server_context.minimum_version = ssl.TLSVersion.TLSv1_2 server_context.maximum_version = ssl.TLSVersion.TLSv1_2 @@ -3755,10 +3853,8 @@ class ThreadedTests(unittest.TestCase): s.connect((HOST, server.port)) self.assertIn("alert", str(e.exception)) - - @unittest.skipUnless(hasattr(ssl.SSLContext, 'minimum_version'), - "required OpenSSL 1.1.0g") - @unittest.skipUnless(ssl.HAS_SSLv3, "requires SSLv3 support") + @requires_minimum_version + @requires_tls_version('SSLv3') def test_min_max_version_sslv3(self): client_context, server_context, hostname = testing_context() server_context.minimum_version = ssl.TLSVersion.SSLv3 @@ -4277,7 +4373,7 @@ class ThreadedTests(unittest.TestCase): 'Session refers to a different SSLContext.') -@unittest.skipUnless(ssl.HAS_TLSv1_3, "Test needs TLS 1.3") +@unittest.skipUnless(has_tls_version('TLSv1_3'), "Test needs TLS 1.3") class TestPostHandshakeAuth(unittest.TestCase): def test_pha_setter(self): protocols = [ |