Merge pull request #2 from andir/add_ipv6_pd

added IPv6 Prefix-Delegation
MartijnBraam · Aug 30, 2015 · 7e44d28 · 7e44d28
2 parents 04d46c4 + 21cf3c6
commit 7e44d28
Show file tree

Hide file tree

Showing 3 changed files with 43 additions and 4 deletions.
diff --git a/isc_dhcp_leases/iscdhcpleases.py b/isc_dhcp_leases/iscdhcpleases.py
@@ -23,9 +23,9 @@ def __init__(self, filename):
         self.last_leases = {}
 
         self.regex_leaseblock = re.compile(r"lease (?P<ip>\d+\.\d+\.\d+\.\d+) {(?P<config>[\s\S]+?)\n}")
-        self.regex_leaseblock6 = re.compile(r"ia-(?P<type>ta|na) \"(?P<id>[^\"]+)\" {(?P<config>[\s\S]+?)\n}")
+        self.regex_leaseblock6 = re.compile(r"ia-(?P<type>ta|na|pd) \"(?P<id>[^\"]+)\" {(?P<config>[\s\S]+?)\n}")
         self.regex_properties = re.compile(r"\s+(?P<key>\S+) (?P<value>[\s\S]+?);")
-        self.regex_iaaddr = re.compile(r"iaaddr (?P<ip>[0-9a-f:]+) {(?P<config>[\s\S]+?)\n\s+}")
+        self.regex_iaaddr = re.compile(r"ia(addr|prefix) (?P<ip>[0-9a-f:]+(/[0-9]+)?) {(?P<config>[\s\S]+?)\n\s+}")
 
     def get(self):
         """
@@ -72,7 +72,10 @@ def get_current(self):
         leases = {}
         for lease in all_leases:
             if lease.valid and lease.active:
-                leases[lease.ethernet] = lease
+                if type(lease) is Lease:
+                    leases[lease.ethernet] = lease
+                elif type(lease) is Lease6:
+                    leases['%s-%s' % (lease.type, lease.host_identifier)] = lease
         return leases
 
 

diff --git a/isc_dhcp_leases/test_files/dhcpd6.leases b/isc_dhcp_leases/test_files/dhcpd6.leases
@@ -12,3 +12,12 @@ ia-na "4dv\352\000\001\000\001\035f\037\342\012\000'\000\000\000" {
     ends 2 2015/08/18 17:05:37;
   }
 }
+ia-pd "4dv\352\000\001\000\001\035f\037\342\012\000'\000\000\000" {
+  cltt 2 2015/08/18 16:55:40;
+  iaprefix 2001:610:500:fff::/64 {
+    binding state active;
+    preferred-life 175;
+    max-life 200;
+    ends 2 2015/08/18 17:05:40;
+  }
+}
diff --git a/isc_dhcp_leases/test_iscDhcpLeases.py b/isc_dhcp_leases/test_iscDhcpLeases.py
@@ -37,7 +37,7 @@ def test_get(self):
 
         leases = IscDhcpLeases("isc_dhcp_leases/test_files/dhcpd6.leases")
         result = leases.get()
-        self.assertEqual(len(result), 1)
+        self.assertEqual(len(result), 2)
         self.assertEqual(result[0].ip, "2001:610:600:891d::60")
         self.assertEqual(result[0].host_identifier, "4dv\352\000\001\000\001\035f\037\342\012\000'\000\000\000")
         self.assertEqual(result[0].valid, True)
@@ -48,6 +48,16 @@ def test_get(self):
         self.assertEqual(result[0].last_communication, datetime(2015, 8, 18, 16, 55, 37))
         self.assertEqual(result[0].type, Lease6.NON_TEMPORARY)
 
+        self.assertEqual(result[1].ip, "2001:610:500:fff::/64")
+        self.assertEqual(result[1].host_identifier, "4dv\352\000\001\000\001\035f\037\342\012\000'\000\000\000")
+        self.assertEqual(result[1].valid, True)
+        self.assertEqual(result[1].active, True)
+        self.assertEqual(result[1].binding_state, 'active')
+        self.assertEqual(result[1].preferred_life, 175)
+        self.assertEqual(result[1].max_life, 200)
+        self.assertEqual(result[1].last_communication, datetime(2015, 8, 18, 16, 55, 40))
+        self.assertEqual(result[1].type, Lease6.PREFIX_DELEGATION)
+
     @freeze_time("2015-07-6 8:15:0")
     def test_get_current(self):
         leases = IscDhcpLeases("isc_dhcp_leases/test_files/debian7.leases")
@@ -62,3 +72,20 @@ def test_get_current(self):
         self.assertTrue("64:5a:04:6a:07:a2" in result)
         self.assertTrue(result["14:da:e9:04:c8:a3"].valid)
         self.assertTrue(result["64:5a:04:6a:07:a2"].valid)
+
+
+    def test_get_current_ipv6(self):
+        with freeze_time("2015-08-18 17:0:0"):
+            leases =  IscDhcpLeases("isc_dhcp_leases/test_files/dhcpd6.leases")
+            result = leases.get_current()
+            self.assertEqual(len(result), 2)
+            self.assertTrue("pd-4dv\352\000\001\000\001\035f\037\342\012\000'\000\000\000" in result)
+            self.assertTrue("na-4dv\352\000\001\000\001\035f\037\342\012\000'\000\000\000" in result)
+
+            for key, r in result.items():
+                self.assertTrue(r.valid, key)
+
+        with freeze_time("2015-08-18 18:0:0"):
+            leases =  IscDhcpLeases("isc_dhcp_leases/test_files/dhcpd6.leases")
+            result = leases.get_current()
+            self.assertEqual(len(result), 0)