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)
(cherry picked from commit 250302bf03)
(cherry picked from commit a0cb2a0df5)

octavia/api/drivers/amphora_driver/v1/driver.py

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

octavia/api/drivers/amphora_driver/v2/driver.py

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

octavia/network/data_models.py

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

octavia/network/drivers/neutron/utils.py

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

octavia/network/drivers/noop_driver/driver.py

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

octavia/tests/functional/api/v2/test_load_balancer.py

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

octavia/tests/unit/api/drivers/amphora_driver/v1/test_amphora_driver.py

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

octavia/tests/unit/api/drivers/amphora_driver/v2/test_amphora_driver.py

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

releasenotes/notes/validate-vip-network-params-57662cc3a99f80e5.yaml

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