diff options
Diffstat (limited to 'Lib/test/test_ssl.py')
| -rw-r--r-- | Lib/test/test_ssl.py | 90 |
1 files changed, 73 insertions, 17 deletions
diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py index 03952b1..73d3e3b 100644 --- a/Lib/test/test_ssl.py +++ b/Lib/test/test_ssl.py @@ -1826,6 +1826,7 @@ class SimpleBackgroundTests(unittest.TestCase): s.connect(self.server_addr) cert = s.getpeercert() self.assertTrue(cert) + # Same with a bytes `capath` argument ctx = ssl.SSLContext(ssl.PROTOCOL_TLS) ctx.verify_mode = ssl.CERT_REQUIRED @@ -1841,8 +1842,6 @@ class SimpleBackgroundTests(unittest.TestCase): der = ssl.PEM_cert_to_DER_cert(pem) ctx = ssl.SSLContext(ssl.PROTOCOL_TLS) ctx.verify_mode = ssl.CERT_REQUIRED - # TODO: fix TLSv1.3 support - ctx.options |= ssl.OP_NO_TLSv1_3 ctx.load_verify_locations(cadata=pem) with ctx.wrap_socket(socket.socket(socket.AF_INET)) as s: s.connect(self.server_addr) @@ -1852,8 +1851,6 @@ class SimpleBackgroundTests(unittest.TestCase): # same with DER ctx = ssl.SSLContext(ssl.PROTOCOL_TLS) ctx.verify_mode = ssl.CERT_REQUIRED - # TODO: fix TLSv1.3 support - ctx.options |= ssl.OP_NO_TLSv1_3 ctx.load_verify_locations(cadata=der) with ctx.wrap_socket(socket.socket(socket.AF_INET)) as s: s.connect(self.server_addr) @@ -2109,11 +2106,21 @@ class ThreadedEchoServer(threading.Thread): self.sock, server_side=True) self.server.selected_npn_protocols.append(self.sslconn.selected_npn_protocol()) self.server.selected_alpn_protocols.append(self.sslconn.selected_alpn_protocol()) - except (ssl.SSLError, ConnectionResetError, OSError) as e: + except (ConnectionResetError, BrokenPipeError) as e: # We treat ConnectionResetError as though it were an # SSLError - OpenSSL on Ubuntu abruptly closes the # connection when asked to use an unsupported protocol. # + # BrokenPipeError is raised in TLS 1.3 mode, when OpenSSL + # tries to send session tickets after handshake. + # https://github.com/openssl/openssl/issues/6342 + self.server.conn_errors.append(str(e)) + if self.server.chatty: + handle_error("\n server: bad connection attempt from " + repr(self.addr) + ":\n") + self.running = False + self.close() + return False + except (ssl.SSLError, OSError) as e: # OSError may occur with wrong protocols, e.g. both # sides use PROTOCOL_TLS_SERVER. # @@ -2220,11 +2227,22 @@ class ThreadedEchoServer(threading.Thread): sys.stdout.write(" server: read %r (%s), sending back %r (%s)...\n" % (msg, ctype, msg.lower(), ctype)) self.write(msg.lower()) + except ConnectionResetError: + # XXX: OpenSSL 1.1.1 sometimes raises ConnectionResetError + # when connection is not shut down gracefully. + if self.server.chatty and support.verbose: + sys.stdout.write( + " Connection reset by peer: {}\n".format( + self.addr) + ) + self.close() + self.running = False except OSError: if self.server.chatty: handle_error("Test server failure:\n") self.close() self.running = False + # normally, we'd just stop here, but for the test # harness, we want to stop the server self.server.stop() @@ -2299,6 +2317,11 @@ class ThreadedEchoServer(threading.Thread): pass except KeyboardInterrupt: self.stop() + except BaseException as e: + if support.verbose and self.chatty: + sys.stdout.write( + ' connection handling failed: ' + repr(e) + '\n') + self.sock.close() def stop(self): @@ -2745,8 +2768,6 @@ class ThreadedTests(unittest.TestCase): server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) server_context.load_cert_chain(IDNSANSFILE) - # TODO: fix TLSv1.3 support - server_context.options |= ssl.OP_NO_TLSv1_3 context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) context.verify_mode = ssl.CERT_REQUIRED @@ -2797,7 +2818,7 @@ class ThreadedTests(unittest.TestCase): with self.assertRaises(ssl.CertificateError): s.connect((HOST, server.port)) - def test_wrong_cert(self): + def test_wrong_cert_tls12(self): """Connecting when the server rejects the client's certificate Launch a server with CERT_REQUIRED, and check that trying to @@ -2808,9 +2829,8 @@ class ThreadedTests(unittest.TestCase): client_context.load_cert_chain(WRONG_CERT) # require TLS client authentication server_context.verify_mode = ssl.CERT_REQUIRED - # TODO: fix TLSv1.3 support - # With TLS 1.3, test fails with exception in server thread - server_context.options |= ssl.OP_NO_TLSv1_3 + # TLS 1.3 has different handshake + client_context.maximum_version = ssl.TLSVersion.TLSv1_2 server = ThreadedEchoServer( context=server_context, chatty=True, connectionchatty=True, @@ -2835,6 +2855,36 @@ 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") + def test_wrong_cert_tls13(self): + client_context, server_context, hostname = testing_context() + client_context.load_cert_chain(WRONG_CERT) + server_context.verify_mode = ssl.CERT_REQUIRED + server_context.minimum_version = ssl.TLSVersion.TLSv1_3 + client_context.minimum_version = ssl.TLSVersion.TLSv1_3 + + server = ThreadedEchoServer( + context=server_context, chatty=True, connectionchatty=True, + ) + with server, \ + client_context.wrap_socket(socket.socket(), + server_hostname=hostname) as s: + # TLS 1.3 perform client cert exchange after handshake + s.connect((HOST, server.port)) + try: + s.write(b'data') + s.read(4) + except ssl.SSLError as e: + if support.verbose: + sys.stdout.write("\nSSLError is %r\n" % e) + except OSError as e: + if e.errno != errno.ECONNRESET: + raise + if support.verbose: + sys.stdout.write("\nsocket.error is %r\n" % e) + else: + self.fail("Use of invalid cert should have failed!") + def test_rude_shutdown(self): """A brutal shutdown of an SSL server should raise an OSError in the client when attempting handshake. @@ -3405,7 +3455,7 @@ class ThreadedTests(unittest.TestCase): # Block on the accept and wait on the connection to close. evt.set() remote, peer = server.accept() - remote.recv(1) + remote.send(remote.recv(4)) t = threading.Thread(target=serve) t.start() @@ -3413,6 +3463,8 @@ class ThreadedTests(unittest.TestCase): evt.wait() client = context.wrap_socket(socket.socket()) client.connect((host, port)) + client.send(b'data') + client.recv() client_addr = client.getsockname() client.close() t.join() @@ -3465,7 +3517,7 @@ class ThreadedTests(unittest.TestCase): self.assertIs(s.version(), None) self.assertIs(s._sslobj, None) s.connect((HOST, server.port)) - if ssl.OPENSSL_VERSION_INFO >= (1, 1, 1): + if IS_OPENSSL_1_1_1 and ssl.HAS_TLSv1_3: self.assertEqual(s.version(), 'TLSv1.3') elif ssl.OPENSSL_VERSION_INFO >= (1, 0, 2): self.assertEqual(s.version(), 'TLSv1.2') @@ -3574,8 +3626,6 @@ class ThreadedTests(unittest.TestCase): sys.stdout.write("\n") client_context, server_context, hostname = testing_context() - # TODO: fix TLSv1.3 support - client_context.options |= ssl.OP_NO_TLSv1_3 server = ThreadedEchoServer(context=server_context, chatty=True, @@ -3594,7 +3644,10 @@ class ThreadedTests(unittest.TestCase): # check if it is sane self.assertIsNotNone(cb_data) - self.assertEqual(len(cb_data), 12) # True for TLSv1 + if s.version() == 'TLSv1.3': + self.assertEqual(len(cb_data), 48) + else: + self.assertEqual(len(cb_data), 12) # True for TLSv1 # and compare with the peers version s.write(b"CB tls-unique\n") @@ -3616,7 +3669,10 @@ class ThreadedTests(unittest.TestCase): # is it really unique self.assertNotEqual(cb_data, new_cb_data) self.assertIsNotNone(cb_data) - self.assertEqual(len(cb_data), 12) # True for TLSv1 + if s.version() == 'TLSv1.3': + self.assertEqual(len(cb_data), 48) + else: + self.assertEqual(len(cb_data), 12) # True for TLSv1 s.write(b"CB tls-unique\n") peer_data_repr = s.read().strip() self.assertEqual(peer_data_repr, |