x/vmware-nsx
Code Issues Proposed changes

check subnet overlapping when adding interface to router

Browse Source 
Fixes bug 1046061

Add logic in _check_for_dup_router_subnet to check subnets are not overlapped.
Add testcase test_router_add_interface_overlapped_cidr.

Change-Id: I71b6cce07273686cc5e4636ce640b7201ce00b32
This commit is contained in:
Hisaharu Ishii 2012-09-06 17:32:56 -07:00
parent e6abfc947b
commit 3240d7a456
2 changed files with 47 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
quantum/db/l3_db.py
View File

@ -21,6 +21,7 @@
import logging import logging
import netaddr
import sqlalchemy as sa import sqlalchemy as sa
from sqlalchemy import orm from sqlalchemy import orm
from sqlalchemy.orm import exc from sqlalchemy.orm import exc
@ -215,17 +216,26 @@ class L3_NAT_db_mixin(l3.RouterPluginBase):
rport_qry = context.session.query(models_v2.Port) rport_qry = context.session.query(models_v2.Port)
rports = rport_qry.filter_by( rports = rport_qry.filter_by(
device_id=router_id, device_id=router_id,
device_owner=DEVICE_OWNER_ROUTER_INTF, device_owner=DEVICE_OWNER_ROUTER_INTF,).all()
network_id=network_id).all()
# its possible these ports on on the same network, but # its possible these ports on on the same network, but
# different subnet # different subnet
new_cidr = self._get_subnet(context, subnet_id)['cidr']
new_ipnet = netaddr.IPNetwork(new_cidr)
for p in rports: for p in rports:
for ip in p['fixed_ips']: for ip in p['fixed_ips']:
if ip['subnet_id'] == subnet_id: if ip['subnet_id'] == subnet_id:
msg = ("Router already has a port on subnet %s" msg = ("Router already has a port on subnet %s"
% subnet_id) % subnet_id)
raise q_exc.BadRequest(resource='router', msg=msg) raise q_exc.BadRequest(resource='router', msg=msg)
cidr = self._get_subnet(context, ip['subnet_id'])['cidr']
ipnet = netaddr.IPNetwork(cidr)
match1 = netaddr.all_matching_cidrs(new_ipnet, [cidr])
match2 = netaddr.all_matching_cidrs(ipnet, [new_cidr])
if match1 or match2:
msg = (("Cidr %s of subnet %s is overlapped "
+ "with cidr %s of subnet %s")
% (new_cidr, subnet_id, cidr, ip['subnet_id']))
raise q_exc.BadRequest(resource='router', msg=msg)
except exc.NoResultFound: except exc.NoResultFound:
pass pass

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

@ -401,8 +401,8 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
r['router']['id'], r['router']['id'],
s['subnet']['id'], s['subnet']['id'],
None, None,
expected_code= expected_code=exc.
exc.HTTPBadRequest.code) HTTPBadRequest.code)
body = self._router_interface_action('remove', body = self._router_interface_action('remove',
r['router']['id'], r['router']['id'],
s['subnet']['id'], s['subnet']['id'],
@ -421,22 +421,50 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
r['router']['id'], r['router']['id'],
None, None,
p2['port']['id'], p2['port']['id'],
expected_code= expected_code=exc.
exc.HTTPBadRequest.code) HTTPBadRequest.code)
# clean-up # clean-up
self._router_interface_action('remove', self._router_interface_action('remove',
r['router']['id'], r['router']['id'],
None, None,
p1['port']['id']) p1['port']['id'])
def test_router_add_interface_overlapped_cidr(self):
with self.router() as r:
with self.subnet(cidr='10.0.1.0/24') as s1:
self._router_interface_action('add',
r['router']['id'],
s1['subnet']['id'],
None)
def try_overlapped_cidr(cidr):
with self.subnet(cidr=cidr) as s2:
self._router_interface_action('add',
r['router']['id'],
s2['subnet']['id'],
None,
expected_code=exc.
HTTPBadRequest.code)
# another subnet with same cidr
try_overlapped_cidr('10.0.1.0/24')
# another subnet with overlapped cidr including s1
try_overlapped_cidr('10.0.0.0/16')
# another subnet with overlapped cidr included by s1
try_overlapped_cidr('10.0.1.1/32')
# clean-up
self._router_interface_action('remove',
r['router']['id'],
s1['subnet']['id'],
None)
def test_router_add_interface_no_data(self): def test_router_add_interface_no_data(self):
with self.router() as r: with self.router() as r:
body = self._router_interface_action('add', body = self._router_interface_action('add',
r['router']['id'], r['router']['id'],
None, None,
None, None,
expected_code= expected_code=exc.
exc.HTTPBadRequest.code) HTTPBadRequest.code)
def test_router_add_gateway(self): def test_router_add_gateway(self):
with self.router() as r: with self.router() as r: