Forbid updating vnic type on a bound port

The vnic type should not be changed once the port is bound since it's
related to the actual port binding. The patch validates the port update
operation and fails the update if the vnic type is attempted to be
changed on a bound port.

Closes-bug: #2033090
Change-Id: I5cb79d9da96ba41a7787083c81f522c328fae049
Signed-off-by: Jakub Libosvar <libosvar@redhat.com>

neutron/plugins/ml2/plugin.py

@@ -1849,6 +1849,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):
@@ -1865,8 +1889,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

@@ -2819,6 +2819,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.