openstack
/
octavia
Code Issues Proposed changes

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
This commit is contained in:
Gregory Thiemonge 2020-12-14 18:24:10 +01:00
parent fbbc5f9024
commit dda1d8665c
9 changed files with 87 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
octavia/api/drivers/amphora_driver/v1/driver.py
View File

@ -103,6 +103,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:

7
octavia/api/drivers/amphora_driver/v2/driver.py
View File

@ -105,6 +105,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:

4
octavia/network/data_models.py
View File

@ -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):

3
octavia/network/drivers/neutron/utils.py
View File

@ -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')
)

6
octavia/network/drivers/noop_driver/driver.py
View File

@ -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",

33
octavia/tests/functional/api/v2/test_load_balancer.py
View File

@ -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(),

15
octavia/tests/unit/api/drivers/amphora_driver/v1/test_driver.py
View File

@ -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()

15
octavia/tests/unit/api/drivers/amphora_driver/v2/test_driver.py
View File

@ -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()

5
releasenotes/notes/validate-vip-network-params-57662cc3a99f80e5.yaml Normal file
View File

@ -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.