openstack/octavia-tempest-plugin
Code Issues Proposed changes

Merge "Add ALPN protocol scenario tests"

Browse Source
This commit is contained in:
Zuul 2020-09-22 02:18:04 +00:00 committed by Gerrit Code Review
commit e054755196
3 changed files with 82 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
octavia_tempest_plugin/common/constants.py
View File

@ -67,6 +67,7 @@ DEFAULT_TLS_CONTAINER_REF = 'default_tls_container_ref'
SNI_CONTAINER_REFS = 'sni_container_refs'
DEFAULT_POOL_ID = 'default_pool_id'
L7_POLICIES = 'l7_policies'
ALPN_PROTOCOLS = 'alpn_protocols'
LB_ALGORITHM = 'lb_algorithm'
LB_ALGORITHM_ROUND_ROBIN = 'ROUND_ROBIN'

6
octavia_tempest_plugin/services/load_balancer/v2/listener_client.py
View File

@ -41,6 +41,7 @@ class ListenerClient(base_client.BaseLBaaSClient):
sni_container_refs=Unset, client_authentication=Unset,
client_ca_tls_container_ref=Unset,
client_crl_container_ref=Unset, allowed_cidrs=Unset,
alpn_protocols=Unset,
return_object_only=True):
"""Create a listener.
@ -89,6 +90,8 @@ class ListenerClient(base_client.BaseLBaaSClient):
revocation list file for
TERMINATED_HTTPS listeners.
:param allowed_cidrs: A list of IPv4 or IPv6 CIDRs.
:param alpn_protocols: A list of ALPN protocols for TERMINATED_HTTPS
listeners.
:param return_object_only: If True, the response returns the object
inside the root tag. False returns the full
response from the API.
@ -215,6 +218,7 @@ class ListenerClient(base_client.BaseLBaaSClient):
sni_container_refs=Unset, client_authentication=Unset,
client_ca_tls_container_ref=Unset,
client_crl_container_ref=Unset, allowed_cidrs=Unset,
alpn_protocols=Unset,
return_object_only=True):
"""Update a listener.
@ -261,6 +265,8 @@ class ListenerClient(base_client.BaseLBaaSClient):
revocation list file for
TERMINATED_HTTPS listeners.
:param allowed_cidrs: A list of IPv4 or IPv6 CIDRs.
:param alpn_protocols: A list of ALPN protocols for TERMINATED_HTTPS
listeners.
:param return_object_only: If True, the response returns the object
inside the root tag. False returns the full
response from the API.

75
octavia_tempest_plugin/tests/barbican_scenario/v2/test_tls_barbican.py
View File

@ -15,6 +15,7 @@
import base64
import requests
import socket
import ssl
import tempfile
from cryptography.hazmat.primitives import serialization
@ -1089,3 +1090,77 @@ class TLSWithBarbicanTest(test_base.LoadBalancerBaseTestWithCompute):
LISTENER1_TCP_PORT),
timeout=12, verify=False, cert=(cert_file.name,
key_file.name))
@decorators.idempotent_id('19bade6f-302f-45dc-b316-553f1dfff49c')
def test_alpn_tls_traffic(self):
"""Test ALPN protocol negotiation"""
s_protos = c_protos = ['http/1.1']
expected = 'http/1.1'
self._test_alpn_tls_traffic(s_protos, c_protos, expected)
@decorators.idempotent_id('ee0d15a3-05b7-498d-9b2f-280d4896e597')
def test_alpn_fallback_tls_traffic(self):
"""Test ALPN protocol negotiation fallback"""
s_protos = ['http/1.0', 'http/1.1']
c_protos = ['bogus', 'h2', 'http/1.1']
expected = 'http/1.1'
self._test_alpn_tls_traffic(s_protos, c_protos, expected)
@decorators.idempotent_id('56f4274a-ebd9-42f7-b897-baebc4b8eb5b')
def test_alpn_proto_not_supported_tls_traffic(self):
"""Test failed ALPN protocol negotiation"""
s_protos = ['http/1.1', 'http/1.0']
c_protos = ['h2']
expected = None
self._test_alpn_tls_traffic(s_protos, c_protos, expected)
def _test_alpn_tls_traffic(self, s_protos, c_protos, expected_proto):
"""Test ALPN protocols between client and load balancer.
:param s_protos: ALPN protocols the load balancer accepts during the
SSL/TLS handshake.
:type s_protos: list of str
:param c_protos: ALPN protocols the client advertise during SSL/TLS the
handshake.
:type c_protos: list of str
:param expected_proto: the expected ALPN protocol selected during the
SSL/TLS handshake. Setting to ``None`` means
parties could not agree on ALPN protocol.
:type expected_proto: str
:raises self.skipException: ALPN support not available prior to v2.20.
"""
if not self.mem_listener_client.is_version_supported(
self.api_version, '2.20'):
raise self.skipException('ALPN protocols are only available on '
'Octavia API version 2.20 or newer.')
listener_name = data_utils.rand_name("lb_member_listener1-tls-alpn")
listener_kwargs = {
const.NAME: listener_name,
const.PROTOCOL: const.TERMINATED_HTTPS,
const.PROTOCOL_PORT: '443',
const.LOADBALANCER_ID: self.lb_id,
const.DEFAULT_POOL_ID: self.pool_id,
const.DEFAULT_TLS_CONTAINER_REF: self.server_secret_ref,
const.ALPN_PROTOCOLS: s_protos,
}
listener = self.mem_listener_client.create_listener(**listener_kwargs)
self.listener_id = listener[const.ID]
self.addCleanup(
self.mem_listener_client.cleanup_listener,
self.listener_id,
lb_client=self.mem_lb_client, lb_id=self.lb_id)
waiters.wait_for_status(self.mem_lb_client.show_loadbalancer,
self.lb_id, const.PROVISIONING_STATUS,
const.ACTIVE,
CONF.load_balancer.build_interval,
CONF.load_balancer.build_timeout)
context = ssl.SSLContext(ssl.PROTOCOL_TLS)
context.set_alpn_protocols(c_protos)
s = socket.socket()
ssl_sock = context.wrap_socket(s)
ssl_sock.connect((self.lb_vip_address, 443))
selected_proto = ssl_sock.selected_alpn_protocol()
self.assertEqual(expected_proto, selected_proto)