_validate_network_tenant_ownership must be less strict

Neutron, currently does a strict validation code
so that for non-shared network the subnets and
ports must belong to the same tenant as the network. In
the case of a "service VM" created by
admin user, this function should return thus allowing
admin users to create ports and networks in a tenant
network.

Change-Id: Ied831402d56b98a1323d30eb6a769fd2df5278ee
Closes-Bug: #1221315

diff --git a/neutron/api/v2/base.py b/neutron/api/v2/base.py
index b9cbd19d871a05ed250bb44d820098f5ec73de17..83d842752e17dd7123270de6d7a66cad8e800e77 100644 (file)
--- a/neutron/api/v2/base.py
+++ b/neutron/api/v2/base.py
@@ -604,7 +604,8 @@ class Controller(object):
     def _validate_network_tenant_ownership(self, request, resource_item):
         # TODO(salvatore-orlando): consider whether this check can be folded
         # in the policy engine
-        if self._resource not in ('port', 'subnet'):
+        if (request.context.is_admin or
+                self._resource not in ('port', 'subnet')):
             return
         network = self._plugin.get_network(
             request.context,

diff --git a/neutron/tests/unit/test_db_plugin.py b/neutron/tests/unit/test_db_plugin.py
index 5524002627a0446d1dc83651d2a2e9fbde7c821c..9d7da0621e12f71fffdba6352ecc1c79c33fc8f8 100644 (file)
--- a/neutron/tests/unit/test_db_plugin.py
+++ b/neutron/tests/unit/test_db_plugin.py
@@ -794,18 +794,27 @@ class TestPortsV2(NeutronDbPluginV2TestCase):
             self.assertEqual(ips[0]['ip_address'], '10.0.0.2')
             self.assertEqual('myname', port['port']['name'])
 
+    def test_create_port_as_admin(self):
+        with self.network(do_delete=False) as network:
+            self._create_port(self.fmt,
+                              network['network']['id'],
+                              webob.exc.HTTPCreated.code,
+                              tenant_id='bad_tenant_id',
+                              device_id='fake_device',
+                              device_owner='fake_owner',
+                              fixed_ips=[],
+                              set_context=False)
+
     def test_create_port_bad_tenant(self):
         with self.network() as network:
-            data = {'port': {'network_id': network['network']['id'],
-                             'tenant_id': 'bad_tenant_id',
-                             'admin_state_up': True,
-                             'device_id': 'fake_device',
-                             'device_owner': 'fake_owner',
-                             'fixed_ips': []}}
-
-            port_req = self.new_create_request('ports', data)
-            res = port_req.get_response(self.api)
-            self.assertEqual(res.status_int, webob.exc.HTTPForbidden.code)
+            self._create_port(self.fmt,
+                              network['network']['id'],
+                              webob.exc.HTTPNotFound.code,
+                              tenant_id='bad_tenant_id',
+                              device_id='fake_device',
+                              device_owner='fake_owner',
+                              fixed_ips=[],
+                              set_context=True)
 
     def test_create_port_public_network(self):
         keys = [('admin_state_up', True), ('status', self.port_create_status)]
@@ -2484,15 +2493,27 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
 
     def test_create_subnet_bad_tenant(self):
         with self.network() as network:
-            data = {'subnet': {'network_id': network['network']['id'],
-                               'cidr': '10.0.2.0/24',
-                               'ip_version': 4,
-                               'tenant_id': 'bad_tenant_id',
-                               'gateway_ip': '10.0.2.1'}}
+            self._create_subnet(self.fmt,
+                                network['network']['id'],
+                                '10.0.2.0/24',
+                                webob.exc.HTTPNotFound.code,
+                                ip_version=4,
+                                tenant_id='bad_tenant_id',
+                                gateway_ip='10.0.2.1',
+                                device_owner='fake_owner',
+                                set_context=True)
 
-            subnet_req = self.new_create_request('subnets', data)
-            res = subnet_req.get_response(self.api)
-            self.assertEqual(res.status_int, webob.exc.HTTPForbidden.code)
+    def test_create_subnet_as_admin(self):
+        with self.network(do_delete=False) as network:
+            self._create_subnet(self.fmt,
+                                network['network']['id'],
+                                '10.0.2.0/24',
+                                webob.exc.HTTPCreated.code,
+                                ip_version=4,
+                                tenant_id='bad_tenant_id',
+                                gateway_ip='10.0.2.1',
+                                device_owner='fake_owner',
+                                set_context=False)
 
     def test_create_subnet_bad_cidr(self):
         with self.network() as network: