Skip tests on non-ALPN platforms.
diff --git a/OpenSSL/test/test_ssl.py b/OpenSSL/test/test_ssl.py
index 68289a2..b611873 100644
--- a/OpenSSL/test/test_ssl.py
+++ b/OpenSSL/test/test_ssl.py
@@ -42,6 +42,8 @@
from OpenSSL.SSL import (
Context, ContextType, Session, Connection, ConnectionType, SSLeay_version)
+from OpenSSL._util import lib as _lib
+
from OpenSSL.test.util import NON_ASCII, TestCase, b
from OpenSSL.test.test_crypto import (
cleartextCertificatePEM, cleartextPrivateKeyPEM)
@@ -1786,178 +1788,183 @@
"""
Tests for ALPN in PyOpenSSL.
"""
- def test_alpn_success(self):
- """
- Tests that clients and servers that agree on the negotiated ALPN
- protocol can correct establish a connection, and that the agreed
- protocol is reported by the connections.
- """
- select_args = []
- def select(conn, options):
- select_args.append((conn, options))
- return b'spdy/2'
+ # Skip tests on versions that don't support ALPN.
+ if _lib.Cryptography_HAS_ALPN:
- client_context = Context(TLSv1_METHOD)
- client_context.set_alpn_protos([b'http/1.1', b'spdy/2'])
+ def test_alpn_success(self):
+ """
+ Tests that clients and servers that agree on the negotiated ALPN
+ protocol can correct establish a connection, and that the agreed
+ protocol is reported by the connections.
+ """
+ select_args = []
+ def select(conn, options):
+ select_args.append((conn, options))
+ return b'spdy/2'
- server_context = Context(TLSv1_METHOD)
- server_context.set_alpn_select_callback(select)
+ client_context = Context(TLSv1_METHOD)
+ client_context.set_alpn_protos([b'http/1.1', b'spdy/2'])
- # Necessary to actually accept the connection
- server_context.use_privatekey(
- load_privatekey(FILETYPE_PEM, server_key_pem))
- server_context.use_certificate(
- load_certificate(FILETYPE_PEM, server_cert_pem))
+ server_context = Context(TLSv1_METHOD)
+ server_context.set_alpn_select_callback(select)
- # Do a little connection to trigger the logic
- server = Connection(server_context, None)
- server.set_accept_state()
+ # Necessary to actually accept the connection
+ server_context.use_privatekey(
+ load_privatekey(FILETYPE_PEM, server_key_pem))
+ server_context.use_certificate(
+ load_certificate(FILETYPE_PEM, server_cert_pem))
- client = Connection(client_context, None)
- client.set_connect_state()
+ # Do a little connection to trigger the logic
+ server = Connection(server_context, None)
+ server.set_accept_state()
- self._interactInMemory(server, client)
+ client = Connection(client_context, None)
+ client.set_connect_state()
- self.assertEqual([(server, [b'http/1.1', b'spdy/2'])], select_args)
+ self._interactInMemory(server, client)
- self.assertEqual(server.get_alpn_proto_negotiated(), b'spdy/2')
- self.assertEqual(client.get_alpn_proto_negotiated(), b'spdy/2')
+ self.assertEqual([(server, [b'http/1.1', b'spdy/2'])], select_args)
+
+ self.assertEqual(server.get_alpn_proto_negotiated(), b'spdy/2')
+ self.assertEqual(client.get_alpn_proto_negotiated(), b'spdy/2')
- def test_alpn_set_on_connection(self):
- """
- The same as test_alpn_success, but setting the ALPN protocols on the
- connection rather than the context.
- """
- select_args = []
- def select(conn, options):
- select_args.append((conn, options))
- return b'spdy/2'
+ def test_alpn_set_on_connection(self):
+ """
+ The same as test_alpn_success, but setting the ALPN protocols on
+ the connection rather than the context.
+ """
+ select_args = []
+ def select(conn, options):
+ select_args.append((conn, options))
+ return b'spdy/2'
- # Setup the client context but don't set any ALPN protocols.
- client_context = Context(TLSv1_METHOD)
+ # Setup the client context but don't set any ALPN protocols.
+ client_context = Context(TLSv1_METHOD)
- server_context = Context(TLSv1_METHOD)
- server_context.set_alpn_select_callback(select)
+ server_context = Context(TLSv1_METHOD)
+ server_context.set_alpn_select_callback(select)
- # Necessary to actually accept the connection
- server_context.use_privatekey(
- load_privatekey(FILETYPE_PEM, server_key_pem))
- server_context.use_certificate(
- load_certificate(FILETYPE_PEM, server_cert_pem))
+ # Necessary to actually accept the connection
+ server_context.use_privatekey(
+ load_privatekey(FILETYPE_PEM, server_key_pem))
+ server_context.use_certificate(
+ load_certificate(FILETYPE_PEM, server_cert_pem))
- # Do a little connection to trigger the logic
- server = Connection(server_context, None)
- server.set_accept_state()
+ # Do a little connection to trigger the logic
+ server = Connection(server_context, None)
+ server.set_accept_state()
- # Set the ALPN protocols on the client connection.
- client = Connection(client_context, None)
- client.set_alpn_protos([b'http/1.1', b'spdy/2'])
- client.set_connect_state()
+ # Set the ALPN protocols on the client connection.
+ client = Connection(client_context, None)
+ client.set_alpn_protos([b'http/1.1', b'spdy/2'])
+ client.set_connect_state()
- self._interactInMemory(server, client)
+ self._interactInMemory(server, client)
- self.assertEqual([(server, [b'http/1.1', b'spdy/2'])], select_args)
+ self.assertEqual([(server, [b'http/1.1', b'spdy/2'])], select_args)
- self.assertEqual(server.get_alpn_proto_negotiated(), b'spdy/2')
- self.assertEqual(client.get_alpn_proto_negotiated(), b'spdy/2')
+ self.assertEqual(server.get_alpn_proto_negotiated(), b'spdy/2')
+ self.assertEqual(client.get_alpn_proto_negotiated(), b'spdy/2')
- def test_alpn_server_fail(self):
- """
- Tests that when clients and servers cannot agree on what protocol to
- use next that the TLS connection does not get established.
- """
- select_args = []
- def select(conn, options):
- select_args.append((conn, options))
- return b''
+ def test_alpn_server_fail(self):
+ """
+ Tests that when clients and servers cannot agree on what protocol
+ to use next that the TLS connection does not get established.
+ """
+ select_args = []
+ def select(conn, options):
+ select_args.append((conn, options))
+ return b''
- client_context = Context(TLSv1_METHOD)
- client_context.set_alpn_protos([b'http/1.1', b'spdy/2'])
+ client_context = Context(TLSv1_METHOD)
+ client_context.set_alpn_protos([b'http/1.1', b'spdy/2'])
- server_context = Context(TLSv1_METHOD)
- server_context.set_alpn_select_callback(select)
+ server_context = Context(TLSv1_METHOD)
+ server_context.set_alpn_select_callback(select)
- # Necessary to actually accept the connection
- server_context.use_privatekey(
- load_privatekey(FILETYPE_PEM, server_key_pem))
- server_context.use_certificate(
- load_certificate(FILETYPE_PEM, server_cert_pem))
+ # Necessary to actually accept the connection
+ server_context.use_privatekey(
+ load_privatekey(FILETYPE_PEM, server_key_pem))
+ server_context.use_certificate(
+ load_certificate(FILETYPE_PEM, server_cert_pem))
- # Do a little connection to trigger the logic
- server = Connection(server_context, None)
- server.set_accept_state()
+ # Do a little connection to trigger the logic
+ server = Connection(server_context, None)
+ server.set_accept_state()
- client = Connection(client_context, None)
- client.set_connect_state()
+ client = Connection(client_context, None)
+ client.set_connect_state()
- # If the client doesn't return anything, the connection will fail.
- self.assertRaises(Error, self._interactInMemory, server, client)
+ # If the client doesn't return anything, the connection will fail.
+ self.assertRaises(Error, self._interactInMemory, server, client)
- self.assertEqual([(server, [b'http/1.1', b'spdy/2'])], select_args)
+ self.assertEqual([(server, [b'http/1.1', b'spdy/2'])], select_args)
- def test_alpn_no_server(self):
- """
- Tests that when clients and servers cannot agree on what protocol to
- use next because the server doesn't offer ALPN.
- """
- client_context = Context(TLSv1_METHOD)
- client_context.set_alpn_protos([b'http/1.1', b'spdy/2'])
+ def test_alpn_no_server(self):
+ """
+ Tests that when clients and servers cannot agree on what protocol
+ to use next because the server doesn't offer ALPN.
+ """
+ client_context = Context(TLSv1_METHOD)
+ client_context.set_alpn_protos([b'http/1.1', b'spdy/2'])
- server_context = Context(TLSv1_METHOD)
+ server_context = Context(TLSv1_METHOD)
- # Necessary to actually accept the connection
- server_context.use_privatekey(
- load_privatekey(FILETYPE_PEM, server_key_pem))
- server_context.use_certificate(
- load_certificate(FILETYPE_PEM, server_cert_pem))
+ # Necessary to actually accept the connection
+ server_context.use_privatekey(
+ load_privatekey(FILETYPE_PEM, server_key_pem))
+ server_context.use_certificate(
+ load_certificate(FILETYPE_PEM, server_cert_pem))
- # Do a little connection to trigger the logic
- server = Connection(server_context, None)
- server.set_accept_state()
+ # Do a little connection to trigger the logic
+ server = Connection(server_context, None)
+ server.set_accept_state()
- client = Connection(client_context, None)
- client.set_connect_state()
+ client = Connection(client_context, None)
+ client.set_connect_state()
- # Do the dance.
- self._interactInMemory(server, client)
+ # Do the dance.
+ self._interactInMemory(server, client)
- self.assertEqual(client.get_alpn_proto_negotiated(), b'')
+ self.assertEqual(client.get_alpn_proto_negotiated(), b'')
- def test_alpn_callback_exception(self):
- """
- Test that we can handle exceptions in the ALPN select callback.
- """
- select_args = []
- def select(conn, options):
- select_args.append((conn, options))
- raise TypeError
+ def test_alpn_callback_exception(self):
+ """
+ Test that we can handle exceptions in the ALPN select callback.
+ """
+ select_args = []
+ def select(conn, options):
+ select_args.append((conn, options))
+ raise TypeError
- client_context = Context(TLSv1_METHOD)
- client_context.set_alpn_protos([b'http/1.1', b'spdy/2'])
+ client_context = Context(TLSv1_METHOD)
+ client_context.set_alpn_protos([b'http/1.1', b'spdy/2'])
- server_context = Context(TLSv1_METHOD)
- server_context.set_alpn_select_callback(select)
+ server_context = Context(TLSv1_METHOD)
+ server_context.set_alpn_select_callback(select)
- # Necessary to actually accept the connection
- server_context.use_privatekey(
- load_privatekey(FILETYPE_PEM, server_key_pem))
- server_context.use_certificate(
- load_certificate(FILETYPE_PEM, server_cert_pem))
+ # Necessary to actually accept the connection
+ server_context.use_privatekey(
+ load_privatekey(FILETYPE_PEM, server_key_pem))
+ server_context.use_certificate(
+ load_certificate(FILETYPE_PEM, server_cert_pem))
- # Do a little connection to trigger the logic
- server = Connection(server_context, None)
- server.set_accept_state()
+ # Do a little connection to trigger the logic
+ server = Connection(server_context, None)
+ server.set_accept_state()
- client = Connection(client_context, None)
- client.set_connect_state()
+ client = Connection(client_context, None)
+ client.set_connect_state()
- # If the client doesn't return anything, the connection will fail.
- self.assertRaises(TypeError, self._interactInMemory, server, client)
- self.assertEqual([(server, [b'http/1.1', b'spdy/2'])], select_args)
+ # If the client doesn't return anything, the connection will fail.
+ self.assertRaises(
+ TypeError, self._interactInMemory, server, client
+ )
+ self.assertEqual([(server, [b'http/1.1', b'spdy/2'])], select_args)