Add validation for VIP network parameters in amphora driver
Some network parameters can be validated in the API, it would avoid to
handle exceptions in the worker when plugging networking resources.
This commit validates that port_security_enabled is True on the VIP
network when using the amphora driver.
Story: 2008449
Task: 41422
Change-Id: I1236d3c6231a657b2aa53b1e488a4d0fe3215070
(cherry picked from commit dda1d8665c
)
This commit is contained in:
parent
b5c6c884a4
commit
2eff3b21c4
|
@ -23,6 +23,7 @@ from stevedore import driver as stevedore_driver
|
|||
from octavia_lib.api.drivers import data_models as driver_dm
|
||||
from octavia_lib.api.drivers import exceptions
|
||||
from octavia_lib.api.drivers import provider_base as driver_base
|
||||
from octavia_lib.common import constants as lib_consts
|
||||
|
||||
from octavia.api.drivers.amphora_driver import availability_zone_schema
|
||||
from octavia.api.drivers.amphora_driver import flavor_schema
|
||||
|
@ -82,6 +83,13 @@ class AmphoraProviderDriver(driver_base.ProviderDriver):
|
|||
project_id=project_id, vip=vip_obj)
|
||||
|
||||
network_driver = utils.get_network_driver()
|
||||
vip_network = network_driver.get_network(
|
||||
vip_dictionary[lib_consts.VIP_NETWORK_ID])
|
||||
if not vip_network.port_security_enabled:
|
||||
message = "Port security must be enabled on the VIP network."
|
||||
raise exceptions.DriverError(user_fault_string=message,
|
||||
operator_fault_string=message)
|
||||
|
||||
try:
|
||||
vip = network_driver.allocate_vip(lb_obj)
|
||||
except network_base.AllocateVIPException as e:
|
||||
|
|
|
@ -24,6 +24,7 @@ from stevedore import driver as stevedore_driver
|
|||
from octavia_lib.api.drivers import data_models as driver_dm
|
||||
from octavia_lib.api.drivers import exceptions
|
||||
from octavia_lib.api.drivers import provider_base as driver_base
|
||||
from octavia_lib.common import constants as lib_consts
|
||||
|
||||
from octavia.api.drivers.amphora_driver import availability_zone_schema
|
||||
from octavia.api.drivers.amphora_driver import flavor_schema
|
||||
|
@ -84,6 +85,13 @@ class AmphoraProviderDriver(driver_base.ProviderDriver):
|
|||
project_id=project_id, vip=vip_obj)
|
||||
|
||||
network_driver = utils.get_network_driver()
|
||||
vip_network = network_driver.get_network(
|
||||
vip_dictionary[lib_consts.VIP_NETWORK_ID])
|
||||
if not vip_network.port_security_enabled:
|
||||
message = "Port security must be enabled on the VIP network."
|
||||
raise exceptions.DriverError(user_fault_string=message,
|
||||
operator_fault_string=message)
|
||||
|
||||
try:
|
||||
vip = network_driver.allocate_vip(lb_obj)
|
||||
except network_base.AllocateVIPException as e:
|
||||
|
|
|
@ -43,7 +43,8 @@ class Network(data_models.BaseDataModel):
|
|||
provider_network_type=None,
|
||||
provider_physical_network=None,
|
||||
provider_segmentation_id=None,
|
||||
router_external=None):
|
||||
router_external=None,
|
||||
port_security_enabled=None):
|
||||
self.id = id
|
||||
self.name = name
|
||||
self.subnets = subnets
|
||||
|
@ -54,6 +55,7 @@ class Network(data_models.BaseDataModel):
|
|||
self.provider_segmentation_id = provider_segmentation_id
|
||||
self.router_external = router_external
|
||||
self.mtu = mtu
|
||||
self.port_security_enabled = port_security_enabled
|
||||
|
||||
|
||||
class Subnet(data_models.BaseDataModel):
|
||||
|
|
|
@ -67,7 +67,8 @@ def convert_network_dict_to_model(network_dict):
|
|||
provider_network_type=nw.get('provider:network_type'),
|
||||
provider_physical_network=nw.get('provider:physical_network'),
|
||||
provider_segmentation_id=nw.get('provider:segmentation_id'),
|
||||
router_external=nw.get('router:external')
|
||||
router_external=nw.get('router:external'),
|
||||
port_security_enabled=nw.get('port_security_enabled')
|
||||
)
|
||||
|
||||
|
||||
|
|
|
@ -149,7 +149,8 @@ class NoopManager(object):
|
|||
LOG.debug("Network %s no-op, get_network network_id %s",
|
||||
self.__class__.__name__, network_id)
|
||||
self.networkconfigconfig[network_id] = (network_id, 'get_network')
|
||||
network = network_models.Network(id=uuidutils.generate_uuid())
|
||||
network = network_models.Network(id=uuidutils.generate_uuid(),
|
||||
port_security_enabled=True)
|
||||
|
||||
class ItIsInsideMe(network_models.Subnet):
|
||||
def to_dict(self, **kwargs):
|
||||
|
@ -181,7 +182,8 @@ class NoopManager(object):
|
|||
self.__class__.__name__, network_name)
|
||||
self.networkconfigconfig[network_name] = (network_name,
|
||||
'get_network_by_name')
|
||||
return network_models.Network(id=uuidutils.generate_uuid())
|
||||
return network_models.Network(id=uuidutils.generate_uuid(),
|
||||
port_security_enabled=True)
|
||||
|
||||
def get_subnet_by_name(self, subnet_name):
|
||||
LOG.debug("Subnet %s no-op, get_subnet_by_name subnet_name %s",
|
||||
|
|
|
@ -931,14 +931,39 @@ class TestLoadBalancer(base.BaseAPITest):
|
|||
}
|
||||
lb_json.update(optionals)
|
||||
body = self._build_body(lb_json)
|
||||
with mock.patch('oslo_messaging.get_rpc_transport'):
|
||||
with mock.patch('oslo_messaging.Target'):
|
||||
with mock.patch('oslo_messaging.RPCClient'):
|
||||
response = self.post(self.LBS_PATH, body)
|
||||
with mock.patch(
|
||||
"octavia.network.drivers.noop_driver.driver.NoopManager"
|
||||
".get_network") as mock_get_network, mock.patch(
|
||||
'oslo_messaging.get_rpc_transport'), mock.patch(
|
||||
'oslo_messaging.Target'), mock.patch(
|
||||
'oslo_messaging.RPCClient'):
|
||||
mock_get_network.return_value = mock.MagicMock()
|
||||
mock_get_network.return_value.port_security_enabled = True
|
||||
response = self.post(self.LBS_PATH, body)
|
||||
api_lb = response.json.get(self.root_tag)
|
||||
self._assert_request_matches_response(lb_json, api_lb)
|
||||
return api_lb
|
||||
|
||||
def test_create_provider_octavia_no_port_sec(self, **optionals):
|
||||
lb_json = {'name': 'test1',
|
||||
'vip_subnet_id': uuidutils.generate_uuid(),
|
||||
'project_id': self.project_id,
|
||||
'provider': constants.OCTAVIA
|
||||
}
|
||||
lb_json.update(optionals)
|
||||
body = self._build_body(lb_json)
|
||||
with mock.patch(
|
||||
"octavia.network.drivers.noop_driver.driver.NoopManager"
|
||||
".get_network") as mock_get_network, mock.patch(
|
||||
'oslo_messaging.get_rpc_transport'), mock.patch(
|
||||
'oslo_messaging.Target'), mock.patch(
|
||||
'oslo_messaging.RPCClient'):
|
||||
mock_get_network.return_value = mock.MagicMock()
|
||||
mock_get_network.return_value.port_security_enabled = False
|
||||
response = self.post(self.LBS_PATH, body, status=500)
|
||||
self.assertIn("Port security must be enabled on the VIP network.",
|
||||
response.json.get('faultstring'))
|
||||
|
||||
def test_create_provider_bogus(self, **optionals):
|
||||
lb_json = {'name': 'test1',
|
||||
'vip_subnet_id': uuidutils.generate_uuid(),
|
||||
|
|
|
@ -42,6 +42,21 @@ class TestAmphoraDriver(base.TestRpc):
|
|||
|
||||
self.assertEqual(self.sample_data.provider_vip_dict, provider_vip_dict)
|
||||
|
||||
@mock.patch('octavia.common.utils.get_network_driver')
|
||||
def test_create_vip_port_without_port_security_enabled(
|
||||
self, mock_get_net_driver):
|
||||
mock_net_driver = mock.MagicMock()
|
||||
mock_get_net_driver.return_value = mock_net_driver
|
||||
network = mock.MagicMock()
|
||||
network.port_security_enabled = False
|
||||
mock_net_driver.get_network.return_value = network
|
||||
mock_net_driver.allocate_vip.return_value = self.sample_data.db_vip
|
||||
|
||||
self.assertRaises(exceptions.DriverError,
|
||||
self.amp_driver.create_vip_port,
|
||||
self.sample_data.lb_id, self.sample_data.project_id,
|
||||
self.sample_data.provider_vip_dict)
|
||||
|
||||
@mock.patch('octavia.common.utils.get_network_driver')
|
||||
def test_create_vip_port_failed(self, mock_get_net_driver):
|
||||
mock_net_driver = mock.MagicMock()
|
||||
|
|
|
@ -42,6 +42,21 @@ class TestAmphoraDriver(base.TestRpc):
|
|||
|
||||
self.assertEqual(self.sample_data.provider_vip_dict, provider_vip_dict)
|
||||
|
||||
@mock.patch('octavia.common.utils.get_network_driver')
|
||||
def test_create_vip_port_without_port_security_enabled(
|
||||
self, mock_get_net_driver):
|
||||
mock_net_driver = mock.MagicMock()
|
||||
mock_get_net_driver.return_value = mock_net_driver
|
||||
network = mock.MagicMock()
|
||||
network.port_security_enabled = False
|
||||
mock_net_driver.get_network.return_value = network
|
||||
mock_net_driver.allocate_vip.return_value = self.sample_data.db_vip
|
||||
|
||||
self.assertRaises(exceptions.DriverError,
|
||||
self.amp_driver.create_vip_port,
|
||||
self.sample_data.lb_id, self.sample_data.project_id,
|
||||
self.sample_data.provider_vip_dict)
|
||||
|
||||
@mock.patch('octavia.common.utils.get_network_driver')
|
||||
def test_create_vip_port_failed(self, mock_get_net_driver):
|
||||
mock_net_driver = mock.MagicMock()
|
||||
|
|
|
@ -0,0 +1,5 @@
|
|||
---
|
||||
fixes:
|
||||
- |
|
||||
Add a validation step in the Octavia Amphora driver to ensure that the
|
||||
port_security_enabled parameter is set on the VIP network.
|
Loading…
Reference in New Issue