x/vmware-nsx
Code Issues Proposed changes

Get subnets of router interfaces with an elevated context.

Browse Source 
Fixes bug 1057558

A router may have interfaces owned by other tenants (by admin operations).
An elevated context is required to get subnet information for such interfaces.

Change-Id: Iaf24c842c2c1e3c52573e7f9831d5f6d8fc01885
This commit is contained in:
Akihiro MOTOKI
2012-10-29 22:37:37 +09:00
parent 71be134306
commit c88ee04147
2 changed files with 42 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
quantum/db/l3_db.py
View File

@@ -262,7 +262,8 @@ class L3_NAT_db_mixin(l3.RouterPluginBase):
msg = ("Router already has a port on subnet %s"
% subnet_id)
raise q_exc.BadRequest(resource='router', msg=msg)
cidr = self._get_subnet(context, ip['subnet_id'])['cidr']
cidr = self._get_subnet(context.elevated(),
ip['subnet_id'])['cidr']
ipnet = netaddr.IPNetwork(cidr)
match1 = netaddr.all_matching_cidrs(new_ipnet, [cidr])
match2 = netaddr.all_matching_cidrs(ipnet, [new_cidr])

40
quantum/tests/unit/test_l3_plugin.py
View File

@@ -426,6 +426,46 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
s['subnet']['id'],
None)
def test_router_add_interface_subnet_with_port_from_other_tenant(self):
tenant_id = _uuid()
other_tenant_id = _uuid()
tenant_context = context.Context(user_id=None, tenant_id=tenant_id)
admin_context = context.get_admin_context()
with mock.patch('quantum.context.Context') as ctx:
ctx.return_value = admin_context
with contextlib.nested(
self.router(tenant_id=tenant_id),
self.network(tenant_id=tenant_id),
self.network(tenant_id=other_tenant_id)) as (r, n1, n2):
with contextlib.nested(
self.subnet(network=n1, cidr='10.0.0.0/24'),
self.subnet(network=n2, cidr='10.1.0.0/24')) as (s1, s2):
ctx.return_value = admin_context
body = self._router_interface_action(
'add',
r['router']['id'],
s2['subnet']['id'],
None)
self.assertTrue('port_id' in body)
ctx.return_value = tenant_context
self._router_interface_action(
'add',
r['router']['id'],
s1['subnet']['id'],
None)
self.assertTrue('port_id' in body)
self._router_interface_action(
'remove',
r['router']['id'],
s1['subnet']['id'],
None)
ctx.return_value = admin_context
body = self._router_interface_action(
'remove',
r['router']['id'],
s2['subnet']['id'],
None)
def test_router_add_interface_port(self):
with self.router() as r:
with self.port(no_delete=True) as p: