Merge "[RBAC] Fix setting network as not shared"

neutron/db/db_base_plugin_v2.py

@@ -278,8 +278,23 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
         # raise if multiple tenants found or if the only tenant found
         # is not the owner of the network
         if (len(tenant_ids) > 1 or len(tenant_ids) == 1 and
-            tenant_ids.pop() != original.tenant_id):
-            raise n_exc.InvalidSharedSetting(network=original.name)
+            original.tenant_id not in tenant_ids):
+            self._validate_projects_have_access_to_network(
+                original, tenant_ids)
+
+    def _validate_projects_have_access_to_network(self, network, project_ids):
+        ctx_admin = ctx.get_admin_context()
+        rb_model = rbac_db.NetworkRBAC
+        other_rbac_entries = model_query.query_with_hooks(
+            ctx_admin, rb_model).filter(
+                and_(rb_model.object_id == network.id,
+                     rb_model.action == 'access_as_shared',
+                     rb_model.target_tenant != "*"))
+        allowed_projects = {entry['target_tenant']
+                            for entry in other_rbac_entries}
+        allowed_projects.add(network.project_id)
+        if project_ids - allowed_projects:
+            raise n_exc.InvalidSharedSetting(network=network.name)
 
     def _validate_ipv6_attributes(self, subnet, cur_subnet):
         if cur_subnet:

neutron/tests/unit/db/test_db_base_plugin_v2.py

@@ -2684,6 +2684,38 @@ class TestNetworksV2(NeutronDbPluginV2TestCase):
             port1 = self.deserialize(self.fmt, res1)
             self._delete('ports', port1['port']['id'])
 
+    def test_update_network_set_not_shared_other_tenant_access_via_rbac(self):
+        with self.network(shared=True) as network:
+            ctx = context.get_admin_context()
+            with db_api.context_manager.writer.using(ctx):
+                ctx.session.add(
+                    rbac_db_models.NetworkRBAC(
+                        object_id=network['network']['id'],
+                        action='access_as_shared',
+                        tenant_id=network['network']['tenant_id'],
+                        target_tenant='somebody_else')
+                )
+                ctx.session.add(
+                    rbac_db_models.NetworkRBAC(
+                        object_id=network['network']['id'],
+                        action='access_as_shared',
+                        tenant_id=network['network']['tenant_id'],
+                        target_tenant='one_more_somebody_else')
+                )
+            res1 = self._create_port(self.fmt,
+                                     network['network']['id'],
+                                     webob.exc.HTTPCreated.code,
+                                     tenant_id='somebody_else',
+                                     set_context=True)
+            data = {'network': {'shared': False}}
+            req = self.new_update_request('networks',
+                                          data,
+                                          network['network']['id'])
+            res = self.deserialize(self.fmt, req.get_response(self.api))
+            self.assertFalse(res['network']['shared'])
+            port1 = self.deserialize(self.fmt, res1)
+            self._delete('ports', port1['port']['id'])
+
     def test_update_network_set_not_shared_multi_tenants_returns_409(self):
         with self.network(shared=True) as network:
             res1 = self._create_port(self.fmt,