Merge "Return 409 Conflict to tenant user deleting port attached to FIP" into stable/yoga

Author: Zuul

neutron/db/l3_db.py

@@ -81,6 +81,13 @@
                  'port: %(port_id)s.')
 
 
+# TODO(froyo): Move this exception to neutron-lib as soon as possible, and when
+# a new release is created and pointed to in the requirements remove this code.
+class FipAssociated(n_exc.InUse):
+    message = _('Unable to complete the operation on port "%(port_id)s" '
+                'because the port still has an associated floating IP.')
+
+
 @registry.has_registry_receivers
 class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
                           base_services.WorkerBase,
@@ -1766,12 +1773,27 @@ def disassociate_floatingips(self, context, port_id, do_notify=True):
         @return: set of router-ids that require notification updates
         """
         with db_api.CONTEXT_WRITER.using(context):
+            # NOTE(froyo): Context is elevated to confirm the presence of at
+            # least one FIP associated to the port_id. Additional checks
+            # regarding the tenant's grants will be carried out in following
+            # lines.
             if not l3_obj.FloatingIP.objects_exist(
-                    context, fixed_port_id=port_id):
+                    context.elevated(), fixed_port_id=port_id):
                 return []
 
             floating_ip_objs = l3_obj.FloatingIP.get_objects(
                 context, fixed_port_id=port_id)
+
+            # NOTE(froyo): To ensure that a FIP assigned by an admin user
+            # cannot be disassociated by a tenant user, we raise exception to
+            # generate a 409 Conflict response message that prompts the tenant
+            # user to contact an admin, rather than a 500 error message.
+            if not context.is_admin:
+                floating_ip_objs_admin = l3_obj.FloatingIP.get_objects(
+                    context.elevated(), fixed_port_id=port_id)
+                if floating_ip_objs_admin != floating_ip_objs:
+                    raise FipAssociated(port_id=port_id)
+
             router_ids = {fip.router_id for fip in floating_ip_objs}
             old_fips = {fip.id: self._make_floatingip_dict(fip)
                         for fip in floating_ip_objs}

neutron/tests/unit/db/test_l3_db.py

@@ -328,6 +328,27 @@ def test_prevent_l3_port_existing_floating_ip(self, gp):
 
             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_disassociate_floatingips_conflict_by_fip_attached(self,
+                                                               get_objects,
+                                                               objects_exist):
+        context_tenant = context.Context('tenant', 'tenant', is_admin=False)
+        objects_exist.return_value = True
+        get_objects.side_effect = [
+            [],
+            [{'id': 'floating_ip1', 'port_id': 'port_id'}]]
+        self.assertRaises(l3_db.FipAssociated,
+                          self.db.disassociate_floatingips,
+                          context_tenant,
+                          'port_id')
+        objects_exist.assert_called_once_with(
+            mock.ANY, fixed_port_id='port_id')
+        expected_calls = [
+                mock.call(context_tenant, fixed_port_id='port_id'),
+                mock.call(mock.ANY, fixed_port_id='port_id')]
+        get_objects.assert_has_calls(expected_calls)
+
     @mock.patch.object(directory, 'get_plugin')
     def test_subscribe_address_scope_of_subnetpool(self, gp):
         l3_db.L3RpcNotifierMixin()