x/vmware-nsx
Code Issues Proposed changes

Merge "Ensure tenant owns devices when creating a gateway"

Browse Source
This commit is contained in:
Jenkins 2014-05-19 22:54:44 +00:00 committed by Gerrit Code Review
commit 2930f42877
2 changed files with 50 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
neutron/plugins/vmware/dbexts/networkgw_db.py
View File

@ -259,9 +259,12 @@ class NetworkGatewayMixin(networkgw.NetworkGatewayPluginBase):
tenant_id=tenant_id,
name=gw_data.get('name'))
# Device list is guaranteed to be a valid list
# TODO(salv-orlando): Enforce that gateway device identifiers
# in this list are among the tenant's NSX network gateway devices
# to avoid risk a tenant 'guessing' other tenant's network devices
device_query = self._query_gateway_devices(
context, filters={'id': [device['id']
for device in gw_data['devices']]})
for device in device_query:
if device['tenant_id'] != tenant_id:
raise GatewayDeviceNotFound(device_id=device['id'])
gw_db.devices.extend([NetworkGatewayDeviceReference(**device)
for device in gw_data['devices']])
context.session.add(gw_db)
@ -443,18 +446,25 @@ class NetworkGatewayMixin(networkgw.NetworkGatewayPluginBase):
self._get_gateway_device(context, device_id),
fields, include_nsx_id)
def _query_gateway_devices(self, context,
filters=None, sorts=None,
limit=None, marker=None,
page_reverse=None):
marker_obj = self._get_marker_obj(
context, 'gateway_device', limit, marker)
return self._get_collection_query(context,
NetworkGatewayDevice,
filters=filters,
sorts=sorts,
limit=limit,
marker_obj=marker_obj,
page_reverse=page_reverse)
def get_gateway_devices(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False, include_nsx_id=False):
marker_obj = self._get_marker_obj(
context, 'gateway_device', limit, marker)
query = self._get_collection_query(context,
NetworkGatewayDevice,
filters=filters,
sorts=sorts,
limit=limit,
marker_obj=marker_obj,
page_reverse=page_reverse)
query = self._query_gateway_devices(context, filters, sorts, limit,
marker, page_reverse)
return [self._make_gateway_device_dict(row, fields, include_nsx_id)
for row in query]

35
neutron/tests/unit/vmware/extensions/test_networkgw.py
View File

@ -513,37 +513,58 @@ class NetworkGatewayDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
expected_code=exc.HTTPNotFound.code)
def test_create_network_gateway(self):
tenant_id = _uuid()
with contextlib.nested(
self._gateway_device(name='dev_1'),
self._gateway_device(name='dev_2')) as (dev_1, dev_2):
self._gateway_device(name='dev_1',
tenant_id=tenant_id),
self._gateway_device(name='dev_2',
tenant_id=tenant_id)) as (dev_1, dev_2):
name = 'test-gw'
dev_1_id = dev_1[self.dev_resource]['id']
dev_2_id = dev_2[self.dev_resource]['id']
devices = [{'id': dev_1_id, 'interface_name': 'xxx'},
{'id': dev_2_id, 'interface_name': 'yyy'}]
keys = [('devices', devices), ('name', name)]
with self._network_gateway(name=name, devices=devices) as gw:
with self._network_gateway(name=name,
devices=devices,
tenant_id=tenant_id) as gw:
for k, v in keys:
self.assertEqual(gw[self.gw_resource][k], v)
def test_create_network_gateway_no_interface_name(self):
with self._gateway_device() as dev:
tenant_id = _uuid()
with self._gateway_device(tenant_id=tenant_id) as dev:
name = 'test-gw'
devices = [{'id': dev[self.dev_resource]['id']}]
exp_devices = devices
exp_devices[0]['interface_name'] = 'breth0'
keys = [('devices', exp_devices), ('name', name)]
with self._network_gateway(name=name, devices=devices) as gw:
with self._network_gateway(name=name,
devices=devices,
tenant_id=tenant_id) as gw:
for k, v in keys:
self.assertEqual(gw[self.gw_resource][k], v)
def test_create_network_gateway_not_owned_device_raises_404(self):
# Create a device with a different tenant identifier
with self._gateway_device(name='dev', tenant_id=_uuid()) as dev:
name = 'test-gw'
dev_id = dev[self.dev_resource]['id']
devices = [{'id': dev_id, 'interface_name': 'xxx'}]
res = self._create_network_gateway(
'json', _uuid(), name=name, devices=devices)
self.assertEqual(404, res.status_int)
def test_delete_network_gateway(self):
with self._gateway_device() as dev:
tenant_id = _uuid()
with self._gateway_device(tenant_id=tenant_id) as dev:
name = 'test-gw'
device_id = dev[self.dev_resource]['id']
devices = [{'id': device_id,
'interface_name': 'xxx'}]
with self._network_gateway(name=name, devices=devices) as gw:
with self._network_gateway(name=name,
devices=devices,
tenant_id=tenant_id) as gw:
# Nothing to do here - just let the gateway go
gw_id = gw[self.gw_resource]['id']
# Verify nothing left on db