Merge pull request #106 from Cerberus98/master

Fixes #105

quark/plugin_modules/subnets.py

@@ -96,6 +96,21 @@ def create_subnet(context, subnet):
         _validate_subnet_cidr(context, net_id, sub_attrs["cidr"])
 
         cidr = netaddr.IPNetwork(sub_attrs["cidr"])
+
+        err_vals = {'cidr': sub_attrs["cidr"], 'network_id': net_id}
+        err = _("Requested subnet with cidr: %(cidr)s for "
+                "network: %(network_id)s. Prefix is too small, must be a "
+                "larger subnet. A prefix less than /%(prefix)s is required.")
+
+        if cidr.version == 6 and cidr.prefixlen > 64:
+            err_vals["prefix"] = 65
+            err_msg = err % err_vals
+            raise exceptions.InvalidInput(error_message=err_msg)
+        elif cidr.version == 4 and cidr.prefixlen > 30:
+            err_vals["prefix"] = 31
+            err_msg = err % err_vals
+            raise exceptions.InvalidInput(error_message=err_msg)
+
         gateway_ip = utils.pop_param(sub_attrs, "gateway_ip", str(cidr[1]))
         dns_ips = utils.pop_param(sub_attrs, "dns_nameservers", [])
         host_routes = utils.pop_param(sub_attrs, "host_routes", [])

quark/tests/plugin_modules/test_subnets.py

@@ -276,7 +276,7 @@ class TestQuarkCreateSubnetAllocationPools(test_quark_plugin.TestQuarkPlugin):
         s = dict(subnet=dict(
             allocation_pools=pools,
             ip_version=6,
-            cidr="2607:f0d0:1002:51::0/96",
+            cidr="2607:f0d0:1002:51::0/64",
             network_id=1))
         with self._stubs(s["subnet"]) as (subnet_create):
             resp = self.plugin.create_subnet(self.context, s)
@@ -367,6 +367,52 @@ class TestQuarkCreateSubnet(test_quark_plugin.TestQuarkPlugin):
                               'end': '172.16.0.254'}]
             self.assertEqual(res["allocation_pools"], expected_pools)
 
+    def test_create_subnet_v6_too_small(self):
+        routes = [dict(cidr="0.0.0.0/0", gateway="0.0.0.0")]
+        subnet = dict(
+            subnet=dict(network_id=1,
+                        tenant_id=self.context.tenant_id, ip_version=4,
+                        cidr="1234::/80", gateway_ip="0.0.0.0",
+                        dns_nameservers=neutron_attrs.ATTR_NOT_SPECIFIED,
+                        host_routes=neutron_attrs.ATTR_NOT_SPECIFIED,
+                        enable_dhcp=None))
+        network = dict(network_id=1)
+        with self._stubs(
+            subnet=subnet["subnet"],
+            network=network,
+            routes=routes
+        ) as (subnet_create, dns_create, route_create):
+            dns_nameservers = subnet["subnet"].pop("dns_nameservers")
+            host_routes = subnet["subnet"].pop("host_routes")
+            subnet_request = copy.deepcopy(subnet)
+            subnet_request["subnet"]["dns_nameservers"] = dns_nameservers
+            subnet_request["subnet"]["host_routes"] = host_routes
+            with self.assertRaises(exceptions.InvalidInput):
+                self.plugin.create_subnet(self.context, subnet_request)
+
+    def test_create_subnet_v4_too_small(self):
+        routes = [dict(cidr="0.0.0.0/0", gateway="0.0.0.0")]
+        subnet = dict(
+            subnet=dict(network_id=1,
+                        tenant_id=self.context.tenant_id, ip_version=4,
+                        cidr="192.168.0.0/31", gateway_ip="0.0.0.0",
+                        dns_nameservers=neutron_attrs.ATTR_NOT_SPECIFIED,
+                        host_routes=neutron_attrs.ATTR_NOT_SPECIFIED,
+                        enable_dhcp=None))
+        network = dict(network_id=1)
+        with self._stubs(
+            subnet=subnet["subnet"],
+            network=network,
+            routes=routes
+        ) as (subnet_create, dns_create, route_create):
+            dns_nameservers = subnet["subnet"].pop("dns_nameservers")
+            host_routes = subnet["subnet"].pop("host_routes")
+            subnet_request = copy.deepcopy(subnet)
+            subnet_request["subnet"]["dns_nameservers"] = dns_nameservers
+            subnet_request["subnet"]["host_routes"] = host_routes
+            with self.assertRaises(exceptions.InvalidInput):
+                self.plugin.create_subnet(self.context, subnet_request)
+
     def test_create_subnet_not_admin_segment_id_ignored(self):
         routes = [dict(cidr="0.0.0.0/0", gateway="0.0.0.0")]
         subnet = dict(