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

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

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

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

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

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

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

@@ -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(
-            with mock.patch('oslo_messaging.Target'):
+                "octavia.network.drivers.noop_driver.driver.NoopManager"
-                with mock.patch('oslo_messaging.RPCClient'):
+                ".get_network") as mock_get_network, mock.patch(
-                    response = self.post(self.LBS_PATH, body)
+                '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

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

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

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

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.