Merge pull request #47 from jurraca/bug-fix

util.py: revert leading zeros logic, add is_valid_pfx

kartograf/collectors/parse.py

@@ -29,7 +29,7 @@ def parse_routeviews_pfx2as(context):
                 if context.max_encode and is_out_of_encoding_range(asn, context.max_encode):
                     continue
 
-                if is_bogon_pfx(prefix) or is_bogon_asn(asn):
+                if not prefix or is_bogon_pfx(prefix) or is_bogon_asn(asn):
                     continue
 
                 clean.write(f"{prefix} {asn}\n")
@@ -52,7 +52,7 @@ def parse_routeviews_pfx2as(context):
             prefix = format_pfx(prefix)
             asn = asn.upper().rstrip('\n')
 
-            if is_bogon_pfx(prefix) or is_bogon_asn(asn):
+            if not prefix or is_bogon_pfx(prefix) or is_bogon_asn(asn):
                 continue
 
             if context.max_encode and is_out_of_encoding_range(asn, context.max_encode):

kartograf/irr/parse.py

@@ -70,7 +70,7 @@ def parse_irr(context):
 
                     # Bogon prefixes and ASNs are excluded since they can not
                     # be used for routing.
-                    if is_bogon_pfx(route) or is_bogon_asn(origin):
+                    if not route or is_bogon_pfx(route) or is_bogon_asn(origin):
                         continue
 
                     if context.max_encode and is_out_of_encoding_range(origin, context.max_encode):

kartograf/rpki/parse.py

@@ -60,7 +60,7 @@ def parse_rpki(context):
 
                 # Bogon prefixes and ASNs are excluded since they can not
                 # be used for routing.
-                if is_bogon_pfx(prefix) or is_bogon_asn(asn):
+                if not prefix or is_bogon_pfx(prefix) or is_bogon_asn(asn):
                     continue
 
                 if context.max_encode and is_out_of_encoding_range(asn, context.max_encode):

kartograf/util.py

@@ -133,20 +133,29 @@ def wait_for_launch(wait):
 
 def format_pfx(pfx):
     """
-    We have seen some formatting issues like leading zeros in the prefix,
-    which can cause problems.
+    Attempt to format an IP network or address.
+    If invalid, return None.
     """
-    try:
+    if is_valid_pfx(pfx):
         if "/" in pfx:
-            pattern = r"^0+"
-            match = re.search(pattern, pfx)
-            if match:
-                pfx = re.sub(pattern, "", pfx)
             formatted_pfx = str(ipaddress.ip_network(pfx))
             return f"{formatted_pfx}"
         return str(ipaddress.ip_address(pfx))
+    return None
+
+
+def is_valid_pfx(pfx):
+    """
+    Check whether the IP network or address provided is valid.
+    """
+    try:
+        if "/" in pfx:
+            ipaddress.ip_network(pfx)
+            return True
+        ipaddress.ip_address(pfx)
+        return True
     except ValueError:
-        return pfx
+        return False
 
 
 def get_root_network(pfx):

tests/test_util.py

@@ -1,7 +1,7 @@
-from kartograf.util import format_pfx, get_root_network
+from kartograf.util import format_pfx, is_valid_pfx, get_root_network
 
 def test_valid_ipv4_network():
-    pfx = "192.144.11.0/21"
+    pfx = "192.144.11.0/24"
     assert format_pfx(pfx) == pfx
 
 
@@ -20,28 +20,33 @@ def test_valid_ipv6_addr():
     assert format_pfx(pfx) == pfx
 
 
-def test_ipv4_prefix_with_leading_zeros():
-    pfx = "010.10.00.00/16"
-    expected_output = "10.10.00.00/16"
-    assert format_pfx(pfx) == expected_output
-
-
-def test_ipv6_prefix_with_leading_zeros():
-    pfx = "001:db8::0/24"
-    expected_output = "1:db8::0/24"
-    assert format_pfx(pfx) == expected_output
-
-
 def test_invalid_ip_network():
     pfx = "192.1/asdf"
-    assert format_pfx(pfx) == pfx
+    assert format_pfx(pfx) is None
 
 
 def test_invalid_input():
     pfx = "no.slash"
+    assert format_pfx(pfx) is None
+
+
+def test_private_network():
+    pfx = "0.128.0.0/24"
     assert format_pfx(pfx) == pfx
 
 
+def test_ipv4_prefix_with_leading_zeros():
+    pfx = "010.10.00.00/16"
+    assert format_pfx(pfx) is None
+    assert not is_valid_pfx(pfx)
+
+
+def test_ipv6_prefix_with_leading_zeros():
+    pfx = "001:db8::0/24"
+    assert format_pfx(pfx) is None
+    assert not is_valid_pfx(pfx)
+
+
 def test_get_root_network():
     ipv4 = "192.144.11.0/24"
     assert get_root_network(ipv4) == 192