check subnet overlapping when adding interface to router

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

diff --git a/quantum/db/l3_db.py b/quantum/db/l3_db.py
index e134439cb87f7ccc65c6e77f861ab5f75da3d4b1..cea813d86599f41458fce31353a22a09b27072ef 100644 (file)
--- 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 3ffeb5a911c0fce5ae3ac5b078b1f67b4df8618c..d0bfa4e323c97f43ce30ca85ae8b3039462aa706 100644 (file)
--- 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: