openstack/octavia
Code Issues Proposed changes

Merge "Add validation for VIP network parameters in amphora driver"

Browse Source
This commit is contained in:
Zuul 2021-01-30 00:45:33 +00:00 committed by Gerrit Code Review
commit 2de31e9173
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.