Browse Source

Merge "Add validation for VIP network parameters in amphora driver" into stable/train

changes/49/778949/1
Zuul 1 month ago
committed by Gerrit Code Review
parent
commit
79298be19f
9 changed files with 89 additions and 8 deletions
  1. +8
    -0
      octavia/api/drivers/amphora_driver/v1/driver.py
  2. +8
    -0
      octavia/api/drivers/amphora_driver/v2/driver.py
  3. +3
    -1
      octavia/network/data_models.py
  4. +2
    -1
      octavia/network/drivers/neutron/utils.py
  5. +4
    -2
      octavia/network/drivers/noop_driver/driver.py
  6. +29
    -4
      octavia/tests/functional/api/v2/test_load_balancer.py
  7. +15
    -0
      octavia/tests/unit/api/drivers/amphora_driver/v1/test_amphora_driver.py
  8. +15
    -0
      octavia/tests/unit/api/drivers/amphora_driver/v2/test_amphora_driver.py
  9. +5
    -0
      releasenotes/notes/validate-vip-network-params-57662cc3a99f80e5.yaml

+ 8
- 0
octavia/api/drivers/amphora_driver/v1/driver.py View File

@ -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 flavor_schema
from octavia.api.drivers import utils as driver_utils
@ -68,6 +69,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:


+ 8
- 0
octavia/api/drivers/amphora_driver/v2/driver.py View File

@ -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 flavor_schema
from octavia.api.drivers import utils as driver_utils
@ -67,6 +68,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:


+ 3
- 1
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):


+ 2
- 1
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')
)


+ 4
- 2
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(object):
def __contains__(self, item):
@ -178,7 +179,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",


+ 29
- 4
octavia/tests/functional/api/v2/test_load_balancer.py View File

@ -932,14 +932,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
- 0
octavia/tests/unit/api/drivers/amphora_driver/v1/test_amphora_driver.py View File

@ -43,6 +43,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
- 0
octavia/tests/unit/api/drivers/amphora_driver/v2/test_amphora_driver.py View File

@ -43,6 +43,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
- 0
releasenotes/notes/validate-vip-network-params-57662cc3a99f80e5.yaml 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.

Loading…
Cancel
Save