Bugfixes in the claim_net algorithm

Author: Johannes Erwerle

minipam/server.py

@@ -72,7 +72,7 @@ def get_net(net, depth=-1):
         c = db_conn.cursor()
         c.execute("SELECT net, address, netmask FROM networks "
                 "WHERE netmask >= ? AND address >= ? AND "
-                "address <= ? ORDER BY netmask, address ASC",
+                "address <= ? ORDER BY address ASC, netmask ASC",
                 (network.prefixlen,
                     network_address_to_int(network),
                     network_broadcast_to_int(network)))
@@ -81,7 +81,7 @@ def get_net(net, depth=-1):
         #    raise_fault("NetworkNotInDatabase")
 
         if len(results) == 0 or results[0]["net"] != str(network):
-            ret = { "address": network.network_address,
+            ret = { "address": str(network.network_address),
                     "cidr" : str(network),
                     "netmask": network.prefixlen,
                     "children": list(),
@@ -177,21 +177,32 @@ def claim_net(net, size):
         network = ip_network(net)
         required_gap = 2**(network.max_prefixlen - size)
 
-        if size > network.max_prefixlen - network.prefixlen:
+        if size < network.prefixlen:
             raise_fault("NoMatchingGapAvailable")
 
         nets = get_net(net, depth=1)
 
         if len(nets["children"]) == 0:
-            add_net(net)
-            return get_net(net, depth=0)
+            network_string = str(network.network_address) + "/" + str(size)
+            add_net(network_string)
+            return get_net(network_string, depth=0)
         else: #at least one child exists.
+
+            def next_start_address(net, size):
+                if size >= net.prefixlen:
+                    start_addr = network_broadcast_to_int(net) + 1
+                else:
+                    size_diff = net.prefixlen - size
+                    supernet = net.supernet(size_diff)
+                    start_addr = int(supernet.network_address) + 2**(supernet.max_prefixlen-supernet.prefixlen)
+                return start_addr
+
             smallest_gap = None
+
             # start and end gaps
             first_block = ip_network(nets["children"][0]["cidr"])
-            last_block = ip_network(nets["children"][-1]["cidr"])
             first_gap = network_address_to_int(first_block) - network_address_to_int(network)
-            last_gap =  network_broadcast_to_int(network) - network_broadcast_to_int(last_block) -1
+
             if first_gap >= required_gap:
                 smallest_gap = { "address" : network.network_address,
                         "length" : first_gap }
@@ -200,12 +211,18 @@ def claim_net(net, size):
             for i, n in enumerate(nets["children"][:-1]):
                 net1 = ip_network(n["cidr"])
                 net2 = ip_network(nets["children"][i+1]["cidr"])
-                gap = network_address_to_int(net2) - network_broadcast_to_int(net1) - 1
+                start_addr = next_start_address(net1, size)
+                gap = network_address_to_int(net2) - start_addr
                 if gap >= required_gap and (smallest_gap is None or gap < smallest_gap["length"]):
-                    smallest_gap = {"address" : ip_address(network_broadcast_to_int(net1) + 1),
+                    smallest_gap = {"address" : ip_address(start_addr),
                             "length": gap}
-            if last_gap > required_gap and (smallest_gap is None or last_gap < smallest_gap["length"]):
-                smallest_gap = { "address" : ip_address(network_broadcast_to_int(last_block) + 1),
+
+            # the last gap
+            last_block = ip_network(nets["children"][-1]["cidr"])
+            last_gap_start = next_start_address(last_block, size)
+            last_gap =  network_broadcast_to_int(network) - last_gap_start + 1
+            if last_gap >= required_gap and (smallest_gap is None or last_gap < smallest_gap["length"]):
+                smallest_gap = { "address" : ip_address(last_gap_start),
                         "length" :last_gap}
 
             if smallest_gap is None:

run.py

@@ -2,9 +2,6 @@
 
 if __name__ == "__main__":
     start_database_connection()
-    add_net("192.168.0.0/16")
-    add_tag("192.168.0.0/16", "name", "lan network")
-    add_tag("192.168.0.0/16", "name", "lan network2")
     setup_xmlrpc_server()
 
 

test/net_methods.py

@@ -159,6 +159,15 @@ def tearDown(self):
         close_database_connection()
         remove("test.db")
 
+    #@unittest.skip("")
+    def test_claim_normal_empty_subnet(self):
+        claimed = claim_net("127.0.0.0/8", 16)
+        result = get_net("127.0.0.0/8")
+        self.assertEqual(claimed["cidr"], "127.0.0.0/16")
+        self.assertEqual(len(result["children"]), 1)
+        self.assertEqual(result["children"][0]["cidr"], "127.0.0.0/16")
+
+    #@unittest.skip("")
     def test_claim_normal(self):
         add_net("127.0.0.0/16")
         add_net("127.2.0.0/16")
@@ -170,6 +179,7 @@ def test_claim_normal(self):
         for i in range(3):
             self.assertEqual(result["children"][i]["address"], "127.%d.0.0" % i)
 
+    #@unittest.skip("")
     def test_claim_first_gap(self):
         add_net("127.1.0.0/16")
         add_net("127.2.0.0/16")
@@ -180,6 +190,7 @@ def test_claim_first_gap(self):
         for i in range(3):
             self.assertEqual(result["children"][i]["address"], "127.%d.0.0" % i)
 
+    #@unittest.skip("")
     def test_claim_last_gap(self):
         add_net("127.0.0.0/16")
         add_net("127.1.0.0/16")
@@ -190,6 +201,25 @@ def test_claim_last_gap(self):
         for i in range(3):
             self.assertEqual(result["children"][i]["address"], "127.%d.0.0" % i)
 
+    #@unittest.skip("")
+    def test_claim_only_last_gap_remaining(self):
+        add_net("127.0.0.0/16")
+        add_net("127.0.0.0/18")
+        add_net("127.0.64.0/18")
+        add_net("127.0.128.0/18")
+        claim_net("127.0.0.0/16", 18)
+        result = get_net("127.0.0.0/16")
+        self.assertEqual(result["cidr"] , "127.0.0.0/16")
+        self.assertEqual(len(result["children"]), 4)
+        for i in range(4):
+            self.assertEqual(result["children"][i]["address"], "127.0.%s.0" % str(i*64))
+
+    def test_claim_last_gap_with_offset(self):
+        add_net("1.0.0.0/8")
+        add_net("1.2.3.0/24")
+        result = claim_net("1.0.0.0/8", 9)
+        self.assertEqual(result["cidr"], "1.128.0.0/9")
+
     def test_claim_no_matching_gap(self):
         add_net("127.0.0.0/9")
         add_net("127.128.0.0/9")
@@ -201,3 +231,11 @@ def test_claim_host_bits_set(self):
         with self.assertRaises(Fault) as cm:
             claim_net("127.0.0.1/8", 16)
         self.assertEqual(cm.exception.faultString, "InvalidNetworkDescription")
+
+    def test_claim_no_matching_gap_at_start(self):
+        add_net("1.0.0.0/8")
+        add_net("1.2.3.0/24")
+        add_net("1.128.0.0/9")
+        with self.assertRaises(Fault) as cm:
+            claim_net("1.0.0.0/8", 9)
+        self.assertEqual(cm.exception.faultString, "NoMatchingGapAvailable")