diff --git a/quantum/db/l3_db.py b/quantum/db/l3_db.py index e134439cb8..cea813d865 100644 --- a/quantum/db/l3_db.py +++ b/quantum/db/l3_db.py @@ -21,6 +21,7 @@ import logging +import netaddr import sqlalchemy as sa from sqlalchemy import orm from sqlalchemy.orm import exc @@ -215,17 +216,26 @@ class L3_NAT_db_mixin(l3.RouterPluginBase): rport_qry = context.session.query(models_v2.Port) rports = rport_qry.filter_by( device_id=router_id, - device_owner=DEVICE_OWNER_ROUTER_INTF, - network_id=network_id).all() + device_owner=DEVICE_OWNER_ROUTER_INTF,).all() # its possible these ports on on the same network, but # different subnet + new_cidr = self._get_subnet(context, subnet_id)['cidr'] + new_ipnet = netaddr.IPNetwork(new_cidr) for p in rports: for ip in p['fixed_ips']: if ip['subnet_id'] == subnet_id: 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'] + 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: pass diff --git a/quantum/tests/unit/test_l3_plugin.py b/quantum/tests/unit/test_l3_plugin.py index 3ffeb5a911..d0bfa4e323 100644 --- a/quantum/tests/unit/test_l3_plugin.py +++ b/quantum/tests/unit/test_l3_plugin.py @@ -401,8 +401,8 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase): r['router']['id'], s['subnet']['id'], None, - expected_code= - exc.HTTPBadRequest.code) + expected_code=exc. + HTTPBadRequest.code) body = self._router_interface_action('remove', r['router']['id'], s['subnet']['id'], @@ -421,22 +421,50 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase): r['router']['id'], None, p2['port']['id'], - expected_code= - exc.HTTPBadRequest.code) + expected_code=exc. + HTTPBadRequest.code) # clean-up self._router_interface_action('remove', r['router']['id'], None, 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): with self.router() as r: body = self._router_interface_action('add', r['router']['id'], None, None, - expected_code= - exc.HTTPBadRequest.code) + expected_code=exc. + HTTPBadRequest.code) def test_router_add_gateway(self): with self.router() as r: