From 7aa941cc09aef8efe54f5bac111248d296e9c8ef Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?S=C5=82awek=20Kap=C5=82o=C5=84ski?= <slawek@kaplonski.pl>
Date: Mon, 16 Apr 2018 13:17:17 +0200
Subject: [PATCH] [RBAC] Fix setting network as not shared

In case when network was shared with specified project
by RBAC rule and it was also set as "shared" there was
a bug which forbid to set such network as not shared even
if only projects which still used network was owner and
project with specified RBAC rule.

This patch fixes it by adding additional check in
NeutronDbPluginV2._validate_shared_update() in such case.

Change-Id: I6ab05a8f0ece454f5bef8ba978af05f5fa1354d8
Closes-Bug: #1764330
---
 neutron/db/db_base_plugin_v2.py               | 19 +++++++++--
 .../tests/unit/db/test_db_base_plugin_v2.py   | 32 +++++++++++++++++++
 2 files changed, 49 insertions(+), 2 deletions(-)

diff --git a/neutron/db/db_base_plugin_v2.py b/neutron/db/db_base_plugin_v2.py
index d471a574502..70f3b39787d 100644
--- a/neutron/db/db_base_plugin_v2.py
+++ b/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:
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 f6240cb5efb..67542b3ea53 100644
--- a/neutron/tests/unit/db/test_db_base_plugin_v2.py
+++ b/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,