_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
This commit is contained in:
Avishay Balderman 2013-09-11 13:46:56 +02:00
parent 03858476f1
commit abe3089906
2 changed files with 41 additions and 19 deletions

View File

@ -604,7 +604,8 @@ class Controller(object):
def _validate_network_tenant_ownership(self, request, resource_item): def _validate_network_tenant_ownership(self, request, resource_item):
# TODO(salvatore-orlando): consider whether this check can be folded # TODO(salvatore-orlando): consider whether this check can be folded
# in the policy engine # in the policy engine
if self._resource not in ('port', 'subnet'): if (request.context.is_admin or
self._resource not in ('port', 'subnet')):
return return
network = self._plugin.get_network( network = self._plugin.get_network(
request.context, request.context,

View File

@ -794,18 +794,27 @@ class TestPortsV2(NeutronDbPluginV2TestCase):
self.assertEqual(ips[0]['ip_address'], '10.0.0.2') self.assertEqual(ips[0]['ip_address'], '10.0.0.2')
self.assertEqual('myname', port['port']['name']) 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): def test_create_port_bad_tenant(self):
with self.network() as network: with self.network() as network:
data = {'port': {'network_id': network['network']['id'], self._create_port(self.fmt,
'tenant_id': 'bad_tenant_id', network['network']['id'],
'admin_state_up': True, webob.exc.HTTPNotFound.code,
'device_id': 'fake_device', tenant_id='bad_tenant_id',
'device_owner': 'fake_owner', device_id='fake_device',
'fixed_ips': []}} device_owner='fake_owner',
fixed_ips=[],
port_req = self.new_create_request('ports', data) set_context=True)
res = port_req.get_response(self.api)
self.assertEqual(res.status_int, webob.exc.HTTPForbidden.code)
def test_create_port_public_network(self): def test_create_port_public_network(self):
keys = [('admin_state_up', True), ('status', self.port_create_status)] keys = [('admin_state_up', True), ('status', self.port_create_status)]
@ -2484,15 +2493,27 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
def test_create_subnet_bad_tenant(self): def test_create_subnet_bad_tenant(self):
with self.network() as network: with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'], self._create_subnet(self.fmt,
'cidr': '10.0.2.0/24', network['network']['id'],
'ip_version': 4, '10.0.2.0/24',
'tenant_id': 'bad_tenant_id', webob.exc.HTTPNotFound.code,
'gateway_ip': '10.0.2.1'}} 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) def test_create_subnet_as_admin(self):
res = subnet_req.get_response(self.api) with self.network(do_delete=False) as network:
self.assertEqual(res.status_int, webob.exc.HTTPForbidden.code) 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): def test_create_subnet_bad_cidr(self):
with self.network() as network: with self.network() as network: