From abe30899065d39b782ccdaf583b806c4a934a504 Mon Sep 17 00:00:00 2001 From: Avishay Balderman Date: Wed, 11 Sep 2013 13:46:56 +0200 Subject: [PATCH] _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 --- neutron/api/v2/base.py | 3 +- neutron/tests/unit/test_db_plugin.py | 57 +++++++++++++++++++--------- 2 files changed, 41 insertions(+), 19 deletions(-) diff --git a/neutron/api/v2/base.py b/neutron/api/v2/base.py index b9cbd19d87..83d842752e 100644 --- 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 5524002627..9d7da0621e 100644 --- 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: