Prevent internal IP change for floating IP

Raise an error when deleting/changing the fixed IP
which is linked to a floating IP.

Closes-Bug: #1999209
Change-Id: I83a5b6c30d54435426f75f4cd1f80bf41822eec5

neutron/db/l3_db.py

@@ -122,6 +122,18 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
                 payload.context, payload.resource_id,
                 port=payload.metadata.get('port'))
 
+    @staticmethod
+    @registry.receives(resources.PORT, [events.BEFORE_UPDATE])
+    def _prevent_internal_ip_change_for_fip(resource, event,
+                                            trigger, payload=None):
+        l3plugin = directory.get_plugin(plugin_constants.L3)
+        new_port = payload.states[1]
+        if (l3plugin and payload.metadata and
+                payload.metadata.get('fixed_ips_updated', False)):
+            l3plugin.prevent_internal_ip_change_for_fip(
+                payload.context, payload.resource_id,
+                new_port['fixed_ips'])
+
     @staticmethod
     def _validate_subnet_address_mode(subnet):
         if (subnet['ip_version'] == 6 and subnet['ipv6_ra_mode'] is None and
@@ -1729,6 +1741,21 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
         filters = filters or {}
         return l3_obj.FloatingIP.count(context, **filters)
 
+    def prevent_internal_ip_change_for_fip(self, context, port_id,
+                                           new_fixed_ips):
+        fips = self._get_floatingips_by_port_id(context, port_id)
+        if not fips or not fips[0].fixed_ip_address:
+            return
+        internal_ip = str(fips[0].fixed_ip_address)
+        for fixed_ip in new_fixed_ips:
+            if fixed_ip.get('ip_address') == internal_ip:
+                return
+        msg = (_('Cannot update the fixed_ips of the port %s, because '
+                 'its original fixed_ip has been associated to a '
+                 'floating ip') %
+               port_id)
+        raise n_exc.BadRequest(resource='port', msg=msg)
+
     def prevent_l3_port_deletion(self, context, port_id, port=None):
         """Checks to make sure a port is allowed to be deleted.
 

neutron/plugins/ml2/plugin.py

@@ -1856,10 +1856,12 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
         need_port_update_notify = False
         bound_mech_contexts = []
         original_port = self.get_port(context, id)
+        metadata = {'fixed_ips_updated': bool('fixed_ips' in attrs)}
         registry.publish(resources.PORT, events.BEFORE_UPDATE, self,
                          payload=events.DBEventPayload(
                              context,
                              resource_id=id,
+                             metadata=metadata,
                              states=(original_port, attrs)))
         with db_api.CONTEXT_WRITER.using(context):
             port_db = self._get_port(context, id)

neutron/tests/unit/db/test_l3_db.py

@@ -308,6 +308,25 @@ class TestL3_NAT_dbonly_mixin(
 
             self.db.prevent_l3_port_deletion(ctx, None)
 
+    @mock.patch.object(l3_obj.FloatingIP, 'objects_exist')
+    @mock.patch.object(l3_obj.FloatingIP, 'get_objects')
+    def test_prevent_internal_ip_change_for_fip(self,
+                                                get_objects,
+                                                objects_exist):
+        ctx = context.get_admin_context()
+        port_id = 'test_internal_port'
+        new_fixed_ips = [{'subnet_id': 'test_subnet',
+                          'ip_address': '192.168.2.110'}]
+        fip_obj_dict = {'fixed_ip_address': '192.168.2.120',
+                        'id': 'floating_ip1',
+                        'port_id': port_id}
+        fip_obj = mock.Mock(**fip_obj_dict)
+        objects_exist.return_value = True
+        get_objects.return_value = [fip_obj]
+        with testtools.ExpectedException(n_exc.BadRequest):
+            self.db.prevent_internal_ip_change_for_fip(ctx, port_id,
+                                                       new_fixed_ips)
+
     @mock.patch.object(l3_obj.FloatingIP, 'objects_exist')
     @mock.patch.object(l3_obj.FloatingIP, 'get_objects')
     def test_disassociate_floatingips_conflict_by_fip_attached(self,

releasenotes/notes/bug-1999209-febf1fa3512556b3.yaml

@@ -0,0 +1,7 @@
+---
+fixes:
+  - |
+    [`bug 1999209 <https://bugs.launchpad.net/neutron/+bug/1999209>`_]
+    Neutron-API now prevents internal IP change for floating IP. It
+    will raise an error when deleting/changing the fixed IP which is
+    linked to a floating IP.