487 lines
21 KiB
Python
487 lines
21 KiB
Python
# Copyright 2016 Blue Box, an IBM Company
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
from unittest import mock
|
|
|
|
from oslo_config import cfg
|
|
from oslo_config import fixture as oslo_fixture
|
|
from oslo_utils import uuidutils
|
|
|
|
import octavia.common.constants as constants
|
|
import octavia.common.exceptions as exceptions
|
|
import octavia.common.validate as validate
|
|
from octavia.network import base as network_base
|
|
from octavia.network import data_models as network_models
|
|
import octavia.tests.unit.base as base
|
|
|
|
|
|
class TestValidations(base.TestCase):
|
|
# Note that particularly complex validation testing is handled via
|
|
# functional tests elsewhere (ex. repository tests)
|
|
|
|
def setUp(self):
|
|
super(TestValidations, self).setUp()
|
|
self.conf = oslo_fixture.Config(cfg.CONF)
|
|
|
|
def test_validate_url(self):
|
|
ret = validate.url('http://example.com')
|
|
self.assertTrue(ret)
|
|
|
|
def test_validate_bad_url(self):
|
|
self.assertRaises(exceptions.InvalidURL, validate.url, 'bad url')
|
|
|
|
def test_validate_url_bad_schema(self):
|
|
self.assertRaises(exceptions.InvalidURL, validate.url,
|
|
'ssh://www.example.com/')
|
|
|
|
def test_validate_header_name(self):
|
|
ret = validate.header_name('Some-header')
|
|
self.assertTrue(ret)
|
|
|
|
def test_validate_bad_header_name(self):
|
|
self.assertRaises(exceptions.InvalidString,
|
|
validate.cookie_value_string,
|
|
'bad header')
|
|
|
|
def test_validate_cookie_value_string(self):
|
|
ret = validate.cookie_value_string('some-cookie')
|
|
self.assertTrue(ret)
|
|
|
|
def test_validate_bad_cookie_value_string(self):
|
|
self.assertRaises(exceptions.InvalidString,
|
|
validate.cookie_value_string,
|
|
'bad cookie value;')
|
|
|
|
def test_validate_header_value_string(self):
|
|
ret = validate.header_value_string('some-value')
|
|
self.assertTrue(ret)
|
|
|
|
def test_validate_header_value_string_quoted(self):
|
|
ret = validate.header_value_string('"some value"')
|
|
self.assertTrue(ret)
|
|
|
|
def test_validate_bad_header_value_string(self):
|
|
self.assertRaises(exceptions.InvalidString,
|
|
validate.header_value_string,
|
|
'\x18')
|
|
|
|
def test_validate_regex(self):
|
|
ret = validate.regex('some regex.*')
|
|
self.assertTrue(ret)
|
|
|
|
def test_validate_bad_regex(self):
|
|
self.assertRaises(exceptions.InvalidRegex, validate.regex,
|
|
'bad regex\\')
|
|
|
|
def test_sanitize_l7policy_api_args_action_reject(self):
|
|
l7p = {'action': constants.L7POLICY_ACTION_REJECT,
|
|
'redirect_url': 'http://www.example.com/',
|
|
'redirect_pool_id': 'test-pool',
|
|
'redirect_pool': {
|
|
'protocol': constants.PROTOCOL_HTTP,
|
|
'lb_algorithm': constants.LB_ALGORITHM_ROUND_ROBIN}}
|
|
s_l7p = validate.sanitize_l7policy_api_args(l7p)
|
|
self.assertIsNone(s_l7p['redirect_url'])
|
|
self.assertIsNone(s_l7p['redirect_pool_id'])
|
|
self.assertNotIn('redirect_pool', s_l7p.keys())
|
|
|
|
def test_sanitize_l7policy_api_args_action_rdr_pool_id(self):
|
|
l7p = {'action': constants.L7POLICY_ACTION_REDIRECT_TO_POOL,
|
|
'redirect_url': 'http://www.example.com/',
|
|
'redirect_pool_id': 'test-pool',
|
|
'redirect_pool': {
|
|
'protocol': constants.PROTOCOL_HTTP,
|
|
'lb_algorithm': constants.LB_ALGORITHM_ROUND_ROBIN}}
|
|
s_l7p = validate.sanitize_l7policy_api_args(l7p)
|
|
self.assertIsNone(s_l7p['redirect_url'])
|
|
self.assertNotIn('redirect_pool', s_l7p.keys())
|
|
self.assertIn('redirect_pool_id', s_l7p.keys())
|
|
|
|
def test_sanitize_l7policy_api_args_action_rdr_pool_model(self):
|
|
l7p = {'action': constants.L7POLICY_ACTION_REDIRECT_TO_POOL,
|
|
'redirect_url': 'http://www.example.com/',
|
|
'redirect_pool_id': None,
|
|
'redirect_pool': {
|
|
'protocol': constants.PROTOCOL_HTTP,
|
|
'lb_algorithm': constants.LB_ALGORITHM_ROUND_ROBIN}}
|
|
s_l7p = validate.sanitize_l7policy_api_args(l7p)
|
|
self.assertIsNone(s_l7p['redirect_url'])
|
|
self.assertNotIn('redirect_pool_id', s_l7p.keys())
|
|
self.assertIn('redirect_pool', s_l7p.keys())
|
|
|
|
def test_sanitize_l7policy_api_args_action_rdr_url(self):
|
|
l7p = {'action': constants.L7POLICY_ACTION_REDIRECT_TO_URL,
|
|
'redirect_url': 'http://www.example.com/',
|
|
'redirect_pool_id': 'test-pool',
|
|
'redirect_pool': {
|
|
'protocol': constants.PROTOCOL_HTTP,
|
|
'lb_algorithm': constants.LB_ALGORITHM_ROUND_ROBIN}}
|
|
s_l7p = validate.sanitize_l7policy_api_args(l7p)
|
|
self.assertIn('redirect_url', s_l7p.keys())
|
|
self.assertIsNone(s_l7p['redirect_pool_id'])
|
|
self.assertNotIn('redirect_pool', s_l7p.keys())
|
|
|
|
def test_sanitize_l7policy_api_args_bad_action(self):
|
|
l7p = {'action': 'bad-action',
|
|
'redirect_url': 'http://www.example.com/',
|
|
'redirect_pool_id': 'test-pool',
|
|
'redirect_pool': {
|
|
'protocol': constants.PROTOCOL_HTTP,
|
|
'lb_algorithm': constants.LB_ALGORITHM_ROUND_ROBIN}}
|
|
self.assertRaises(exceptions.InvalidL7PolicyAction,
|
|
validate.sanitize_l7policy_api_args, l7p)
|
|
|
|
def test_sanitize_l7policy_api_args_action_none(self):
|
|
l7p = {'action': None}
|
|
self.assertRaises(exceptions.InvalidL7PolicyAction,
|
|
validate.sanitize_l7policy_api_args, l7p, True)
|
|
|
|
def test_sanitize_l7policy_api_args_both_rdr_args_a(self):
|
|
l7p = {'redirect_url': 'http://www.example.com/',
|
|
'redirect_pool_id': 'test-pool'}
|
|
self.assertRaises(exceptions.InvalidL7PolicyArgs,
|
|
validate.sanitize_l7policy_api_args, l7p)
|
|
|
|
def test_sanitize_l7policy_api_args_both_rdr_args_b(self):
|
|
l7p = {'redirect_url': 'http://www.example.com/',
|
|
'redirect_pool': {
|
|
'protocol': constants.PROTOCOL_HTTP,
|
|
'lb_algorithm': constants.LB_ALGORITHM_ROUND_ROBIN}}
|
|
self.assertRaises(exceptions.InvalidL7PolicyArgs,
|
|
validate.sanitize_l7policy_api_args, l7p)
|
|
|
|
def test_sanitize_l7policy_api_args_rdr_pool_id(self):
|
|
l7p = {'redirect_pool_id': 'test-pool',
|
|
'redirect_url': None,
|
|
'redirect_pool': None}
|
|
s_l7p = validate.sanitize_l7policy_api_args(l7p)
|
|
self.assertIn('redirect_pool_id', s_l7p.keys())
|
|
self.assertIsNone(s_l7p['redirect_url'])
|
|
self.assertNotIn('redirect_pool', s_l7p.keys())
|
|
self.assertIn('action', s_l7p.keys())
|
|
self.assertEqual(constants.L7POLICY_ACTION_REDIRECT_TO_POOL,
|
|
s_l7p['action'])
|
|
|
|
def test_sanitize_l7policy_api_args_rdr_pool_noid(self):
|
|
l7p = {'redirect_pool_id': None,
|
|
'redirect_url': None,
|
|
'redirect_pool': {
|
|
'protocol': constants.PROTOCOL_HTTP,
|
|
'lb_algorithm': constants.LB_ALGORITHM_ROUND_ROBIN}}
|
|
s_l7p = validate.sanitize_l7policy_api_args(l7p)
|
|
self.assertIn('redirect_pool', s_l7p.keys())
|
|
self.assertIsNone(s_l7p['redirect_url'])
|
|
self.assertNotIn('redirect_pool_id', s_l7p.keys())
|
|
self.assertIn('action', s_l7p.keys())
|
|
self.assertEqual(constants.L7POLICY_ACTION_REDIRECT_TO_POOL,
|
|
s_l7p['action'])
|
|
|
|
def test_sanitize_l7policy_api_args_rdr_pool_id_none_create(self):
|
|
l7p = {'redirect_pool_id': None}
|
|
self.assertRaises(exceptions.InvalidL7PolicyAction,
|
|
validate.sanitize_l7policy_api_args, l7p, True)
|
|
|
|
def test_sanitize_l7policy_api_args_rdr_pool_noid_none_create(self):
|
|
l7p = {'redirect_pool': None}
|
|
self.assertRaises(exceptions.InvalidL7PolicyAction,
|
|
validate.sanitize_l7policy_api_args, l7p, True)
|
|
|
|
def test_sanitize_l7policy_api_args_rdr_pool_both_none_create(self):
|
|
l7p = {'redirect_pool': None,
|
|
'redirect_pool_id': None}
|
|
self.assertRaises(exceptions.InvalidL7PolicyAction,
|
|
validate.sanitize_l7policy_api_args, l7p, True)
|
|
|
|
def test_sanitize_l7policy_api_args_rdr_url(self):
|
|
l7p = {'redirect_pool_id': None,
|
|
'redirect_url': 'http://www.example.com/',
|
|
'redirect_pool': None}
|
|
s_l7p = validate.sanitize_l7policy_api_args(l7p)
|
|
self.assertIsNone(s_l7p['redirect_pool_id'])
|
|
self.assertNotIn('redirect_pool', s_l7p.keys())
|
|
self.assertIn('redirect_url', s_l7p.keys())
|
|
self.assertIn('action', s_l7p.keys())
|
|
self.assertEqual(constants.L7POLICY_ACTION_REDIRECT_TO_URL,
|
|
s_l7p['action'])
|
|
|
|
def test_sanitize_l7policy_api_args_rdr_url_none_create(self):
|
|
l7p = {'redirect_url': None}
|
|
self.assertRaises(exceptions.InvalidL7PolicyAction,
|
|
validate.sanitize_l7policy_api_args, l7p, True)
|
|
|
|
def test_sanitize_l7policy_api_args_rdr_url_bad_url(self):
|
|
l7p = {'redirect_url': 'bad url'}
|
|
self.assertRaises(exceptions.InvalidURL,
|
|
validate.sanitize_l7policy_api_args, l7p, True)
|
|
|
|
def test_sanitize_l7policy_api_args_update_action_rdr_pool_arg(self):
|
|
l7p = {'action': constants.L7POLICY_ACTION_REDIRECT_TO_POOL,
|
|
'redirect_url': None,
|
|
'redirect_pool_id': None,
|
|
'redirect_pool': None}
|
|
self.assertRaises(exceptions.InvalidL7PolicyArgs,
|
|
validate.sanitize_l7policy_api_args, l7p)
|
|
|
|
def test_sanitize_l7policy_api_args_update_action_rdr_url_arg(self):
|
|
l7p = {'action': constants.L7POLICY_ACTION_REDIRECT_TO_URL,
|
|
'redirect_url': None,
|
|
'redirect_pool_id': None,
|
|
'redirect_pool': None}
|
|
self.assertRaises(exceptions.InvalidL7PolicyArgs,
|
|
validate.sanitize_l7policy_api_args, l7p)
|
|
|
|
def test_sanitize_l7policy_api_args_create_must_have_action(self):
|
|
l7p = {}
|
|
self.assertRaises(exceptions.InvalidL7PolicyAction,
|
|
validate.sanitize_l7policy_api_args, l7p, True)
|
|
|
|
def test_sanitize_l7policy_api_args_update_must_have_args(self):
|
|
l7p = {}
|
|
self.assertRaises(exceptions.InvalidL7PolicyArgs,
|
|
validate.sanitize_l7policy_api_args, l7p)
|
|
|
|
def test_port_exists_with_bad_port(self):
|
|
port_id = uuidutils.generate_uuid()
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_port = mock.Mock(
|
|
side_effect=network_base.PortNotFound('Port not found'))
|
|
self.assertRaises(
|
|
exceptions.InvalidSubresource,
|
|
validate.port_exists, port_id)
|
|
|
|
def test_port_exists_with_valid_port(self):
|
|
port_id = uuidutils.generate_uuid()
|
|
port = network_models.Port(id=port_id)
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_port.return_value = port
|
|
self.assertEqual(validate.port_exists(port_id), port)
|
|
|
|
def test_check_port_in_use(self):
|
|
port_id = uuidutils.generate_uuid()
|
|
device_id = uuidutils.generate_uuid()
|
|
port = network_models.Port(id=port_id, device_id=device_id)
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_port.device_id = port
|
|
self.assertRaises(
|
|
exceptions.ValidationException,
|
|
validate.check_port_in_use, port)
|
|
|
|
def test_subnet_exists_with_bad_subnet(self):
|
|
subnet_id = uuidutils.generate_uuid()
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_subnet = mock.Mock(
|
|
side_effect=network_base.SubnetNotFound('Subnet not found'))
|
|
self.assertRaises(
|
|
exceptions.InvalidSubresource,
|
|
validate.subnet_exists, subnet_id)
|
|
|
|
def test_subnet_exists_with_valid_subnet(self):
|
|
subnet_id = uuidutils.generate_uuid()
|
|
subnet = network_models.Subnet(id=subnet_id)
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_subnet.return_value = subnet
|
|
self.assertEqual(validate.subnet_exists(subnet_id), subnet)
|
|
|
|
def test_network_exists_with_bad_network(self):
|
|
network_id = uuidutils.generate_uuid()
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_network = mock.Mock(
|
|
side_effect=network_base.NetworkNotFound('Network not found'))
|
|
self.assertRaises(
|
|
exceptions.InvalidSubresource,
|
|
validate.network_exists_optionally_contains_subnet, network_id)
|
|
|
|
def test_network_exists_with_valid_network(self):
|
|
network_id = uuidutils.generate_uuid()
|
|
network = network_models.Network(id=network_id)
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_network.return_value = network
|
|
self.assertEqual(
|
|
validate.network_exists_optionally_contains_subnet(network_id),
|
|
network)
|
|
|
|
def test_network_exists_with_valid_subnet(self):
|
|
network_id = uuidutils.generate_uuid()
|
|
subnet_id = uuidutils.generate_uuid()
|
|
network = network_models.Network(
|
|
id=network_id,
|
|
subnets=[subnet_id])
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_network.return_value = network
|
|
self.assertEqual(
|
|
validate.network_exists_optionally_contains_subnet(
|
|
network_id, subnet_id),
|
|
network)
|
|
|
|
def test_network_exists_with_bad_subnet(self):
|
|
network_id = uuidutils.generate_uuid()
|
|
subnet_id = uuidutils.generate_uuid()
|
|
network = network_models.Network(id=network_id)
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_network.return_value = network
|
|
self.assertRaises(
|
|
exceptions.InvalidSubresource,
|
|
validate.network_exists_optionally_contains_subnet,
|
|
network_id, subnet_id)
|
|
|
|
def test_network_allowed_by_config(self):
|
|
net_id1 = uuidutils.generate_uuid()
|
|
net_id2 = uuidutils.generate_uuid()
|
|
net_id3 = uuidutils.generate_uuid()
|
|
valid_net_ids = ",".join((net_id1, net_id2))
|
|
self.conf.config(group="networking", valid_vip_networks=valid_net_ids)
|
|
validate.network_allowed_by_config(net_id1)
|
|
validate.network_allowed_by_config(net_id2)
|
|
self.assertRaises(
|
|
exceptions.ValidationException,
|
|
validate.network_allowed_by_config, net_id3)
|
|
|
|
def test_qos_policy_exists(self):
|
|
qos_policy_id = uuidutils.generate_uuid()
|
|
qos_policy = network_models.QosPolicy(id=qos_policy_id)
|
|
with mock.patch(
|
|
'octavia.common.utils.get_network_driver') as net_mock:
|
|
net_mock.return_value.get_qos_policy.return_value = qos_policy
|
|
self.assertEqual(
|
|
validate.qos_policy_exists(qos_policy_id),
|
|
qos_policy)
|
|
|
|
net_mock.return_value.get_qos_policy.side_effect = Exception
|
|
self.assertRaises(exceptions.InvalidSubresource,
|
|
validate.qos_policy_exists,
|
|
qos_policy_id)
|
|
|
|
def test_qos_extension_enabled(self):
|
|
network_driver = mock.Mock()
|
|
network_driver.qos_enabled.return_value = True
|
|
self.assertIsNone(validate.qos_extension_enabled(network_driver))
|
|
|
|
def test_qos_extension_disabled(self):
|
|
network_driver = mock.Mock()
|
|
network_driver.qos_enabled.return_value = False
|
|
self.assertRaises(exceptions.ValidationException,
|
|
validate.qos_extension_enabled,
|
|
network_driver)
|
|
|
|
def test_check_session_persistence(self):
|
|
valid_cookie_name_dict = {'type': 'APP_COOKIE',
|
|
'cookie_name': 'chocolate_chip'}
|
|
invalid_cookie_name_dict = {'type': 'APP_COOKIE',
|
|
'cookie_name': '@chocolate_chip'}
|
|
invalid_type_HTTP_cookie_name_dict = {'type': 'HTTP_COOKIE',
|
|
'cookie_name': 'chocolate_chip'}
|
|
invalid_type_IP_cookie_name_dict = {'type': 'SOURCE_IP',
|
|
'cookie_name': 'chocolate_chip'}
|
|
invalid_missing_cookie_name_dict = {'type': 'APP_COOKIE'}
|
|
|
|
# Validate that a good cookie name passes
|
|
validate.check_session_persistence(valid_cookie_name_dict)
|
|
|
|
# Test raises with providing an invalid cookie name
|
|
self.assertRaises(exceptions.ValidationException,
|
|
validate.check_session_persistence,
|
|
invalid_cookie_name_dict)
|
|
|
|
# Test raises type HTTP_COOKIE and providing cookie_name
|
|
self.assertRaises(exceptions.ValidationException,
|
|
validate.check_session_persistence,
|
|
invalid_type_HTTP_cookie_name_dict)
|
|
|
|
# Test raises type SOURCE_IP and providing cookie_name
|
|
self.assertRaises(exceptions.ValidationException,
|
|
validate.check_session_persistence,
|
|
invalid_type_IP_cookie_name_dict)
|
|
|
|
# Test raises when type APP_COOKIE but no cookie_name
|
|
self.assertRaises(exceptions.ValidationException,
|
|
validate.check_session_persistence,
|
|
invalid_missing_cookie_name_dict)
|
|
|
|
# Test catch all exception raises a user friendly message
|
|
with mock.patch('re.compile') as compile_mock:
|
|
compile_mock.side_effect = Exception
|
|
self.assertRaises(exceptions.ValidationException,
|
|
validate.check_session_persistence,
|
|
valid_cookie_name_dict)
|
|
|
|
def test_ip_not_reserved(self):
|
|
self.conf.config(group="networking", reserved_ips=['198.51.100.4'])
|
|
|
|
# Test good address
|
|
validate.ip_not_reserved('203.0.113.5')
|
|
|
|
# Test IPv4 reserved address
|
|
self.assertRaises(exceptions.InvalidOption,
|
|
validate.ip_not_reserved,
|
|
'198.51.100.4')
|
|
|
|
self.conf.config(
|
|
group="networking",
|
|
reserved_ips=['2001:0DB8:0000:0000:0000:0000:0000:0005'])
|
|
|
|
# Test good IPv6 address
|
|
validate.ip_not_reserved('2001:0DB8::9')
|
|
|
|
# Test reserved IPv6 expanded
|
|
self.assertRaises(exceptions.InvalidOption,
|
|
validate.ip_not_reserved,
|
|
'2001:0DB8:0000:0000:0000:0000:0000:0005')
|
|
|
|
# Test reserved IPv6 short hand notation
|
|
self.assertRaises(exceptions.InvalidOption,
|
|
validate.ip_not_reserved,
|
|
'2001:0DB8::5')
|
|
|
|
def test_is_flavor_spares_compatible(self):
|
|
not_compat_flavor = {constants.COMPUTE_FLAVOR: 'chocolate'}
|
|
compat_flavor = {constants.LOADBALANCER_TOPOLOGY:
|
|
constants.TOPOLOGY_SINGLE}
|
|
|
|
self.assertTrue(validate.is_flavor_spares_compatible(None))
|
|
self.assertTrue(validate.is_flavor_spares_compatible(compat_flavor))
|
|
self.assertFalse(
|
|
validate.is_flavor_spares_compatible(not_compat_flavor))
|
|
|
|
def test_check_default_ciphers_blacklist_conflict(self):
|
|
self.conf.config(group='api_settings',
|
|
tls_cipher_blacklist='PSK-AES128-CBC-SHA')
|
|
self.conf.config(group='api_settings',
|
|
default_listener_ciphers='ECDHE-ECDSA-AES256-SHA:'
|
|
'PSK-AES128-CBC-SHA:TLS_AES_256_GCM_SHA384')
|
|
|
|
self.assertRaises(exceptions.ValidationException,
|
|
validate.check_default_ciphers_blacklist_conflict)
|
|
|
|
def test_check_tls_version_list(self):
|
|
# Test valid list
|
|
validate.check_tls_version_list(['TLSv1.1', 'TLSv1.2', 'TLSv1.3'])
|
|
# Test invalid list
|
|
self.assertRaises(
|
|
exceptions.ValidationException,
|
|
validate.check_tls_version_list,
|
|
['SSLv3', 'TLSv1.0'])
|
|
# Test empty list
|
|
self.assertRaises(
|
|
exceptions.ValidationException,
|
|
validate.check_tls_version_list,
|
|
[])
|