diff --git a/neutron/db/db_base_plugin_v2.py b/neutron/db/db_base_plugin_v2.py index f6637eaeaed..dcade09308b 100644 --- a/neutron/db/db_base_plugin_v2.py +++ b/neutron/db/db_base_plugin_v2.py @@ -283,8 +283,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: diff --git a/neutron/tests/unit/db/test_db_base_plugin_v2.py b/neutron/tests/unit/db/test_db_base_plugin_v2.py index 294eecbdece..30b48c2f28c 100644 --- a/neutron/tests/unit/db/test_db_base_plugin_v2.py +++ b/neutron/tests/unit/db/test_db_base_plugin_v2.py @@ -2678,6 +2678,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,