Merge "Forbid updating vnic type on a bound port"

neutron/plugins/ml2/plugin.py

@@ -1851,6 +1851,30 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
             if security_groups:
                 raise psec_exc.PortSecurityPortHasSecurityGroup()
 
+    @staticmethod
+    def _validate_port_update(old_port, new_port, binding):
+        if not binding:
+            raise exc.PortNotFound(port_id=old_port.id)
+        try:
+            old_vnic_type, new_vnic_type = (
+                binding.vnic_type, new_port[portbindings.VNIC_TYPE])
+        except KeyError:
+            return
+
+        if (old_vnic_type != new_vnic_type and
+                binding.vif_type != portbindings.VIF_TYPE_UNBOUND):
+            LOG.info("Attempting to change VNIC TYPE from {old_type} to "
+                     "{new_type} on port {port_id}, this operation is not "
+                     "allowed because the port is bound".format(
+                         old_type=old_vnic_type,
+                         new_type=new_vnic_type,
+                         port_id=old_port.id))
+            raise exc.PortInUse(
+                port_id=old_port.id,
+                net_id=old_port.network_id,
+                device_id=old_port.device_id,
+            )
+
     @utils.transaction_guard
     @db_api.retry_if_session_inactive()
     def update_port(self, context, id, port):
@@ -1869,8 +1893,7 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
             port_db = self._get_port(context, id)
             binding = p_utils.get_port_binding_by_status_and_host(
                 port_db.port_bindings, const.ACTIVE)
-            if not binding:
-                raise exc.PortNotFound(port_id=id)
+            self._validate_port_update(port_db, attrs, binding)
             mac_address_updated = self._check_mac_update_allowed(
                 port_db, attrs, binding)
             mac_address_updated |= self._reset_mac_for_direct_physical(

neutron/tests/unit/plugins/ml2/test_plugin.py

@@ -2851,6 +2851,48 @@ class TestMl2PortBinding(Ml2PluginV2TestCase,
             self.assertTrue(update_mock.mock_calls)
             self.assertEqual('test', binding.host)
 
+    def _test__validate_port_update_prepare(self):
+        plugin, port_context, bound_context = (
+            self._create_port_and_bound_context(
+                portbindings.VIF_TYPE_OVS,
+                portbindings.VIF_TYPE_OVS))
+        port_db = plugin._get_port(self.context, port_context.current['id'])
+
+        return plugin, port_db, port_context._binding
+
+    def test__validate_port_update_no_port_binding(self):
+        plugin, port_db, binding = self._test__validate_port_update_prepare()
+        new_port = {portbindings.VNIC_TYPE: portbindings.VNIC_DIRECT_PHYSICAL}
+
+        with testtools.ExpectedException(exc.PortNotFound):
+            plugin._validate_port_update(port_db, new_port, None)
+
+    def test__validate_port_update_no_vnic_type(self):
+        plugin, port_db, binding = self._test__validate_port_update_prepare()
+        new_port = {portbindings.HOST_ID: 'foo'}
+
+        plugin._validate_port_update(port_db, new_port, binding)
+
+    def test__validate_port_update_vnic_type_unbound_port(self):
+        plugin, port_db, binding = self._test__validate_port_update_prepare()
+        new_port = {portbindings.VNIC_TYPE: portbindings.VNIC_DIRECT_PHYSICAL}
+        binding.vif_type = portbindings.VIF_TYPE_UNBOUND
+
+        plugin._validate_port_update(port_db, new_port, binding)
+
+    def test__validate_port_update_vnic_type_bound_port_same_vnic_type(self):
+        plugin, port_db, binding = self._test__validate_port_update_prepare()
+        new_port = {portbindings.VNIC_TYPE: portbindings.VNIC_NORMAL}
+
+        plugin._validate_port_update(port_db, new_port, binding)
+
+    def test__validate_port_update_vnic_type_bound_port(self):
+        plugin, port_db, binding = self._test__validate_port_update_prepare()
+        new_port = {portbindings.VNIC_TYPE: portbindings.VNIC_DIRECT_PHYSICAL}
+
+        with testtools.ExpectedException(exc.PortInUse):
+            plugin._validate_port_update(port_db, new_port, binding)
+
     def test_process_distributed_port_binding_update_router_id(self):
         host_id = 'host'
         binding = models.DistributedPortBinding(

releasenotes/notes/update-vnic-type-d2cb5b78d5ba1c32.yaml

@@ -0,0 +1,7 @@
+---
+fixes:
+  - |
+    In previous versions, an administrator was allowed to update a port
+    ``binding:vnic_type`` attribute even if it was bound. This is now blocked
+    and the update operation of the attribute returns the ``Conflict (409)``
+    response code.