Clean up issues with IP address parsing in several places

src/zino/snmp.py

@@ -5,6 +5,7 @@
 import threading
 from collections import defaultdict
 from dataclasses import dataclass
+from ipaddress import ip_address
 from typing import Any, NamedTuple, Sequence, Tuple, Union
 
 from pyasn1.type import univ
@@ -509,7 +510,7 @@ def _mib_value_to_python(value: SupportedTypes) -> Union[str, int, OID]:
         value = int(value) if not value.namedValues else value.prettyPrint()
     elif isinstance(value, univ.OctetString):
         if type(value).__name__ in ("InetAddress", "IpAddress"):
-            value = value.prettyPrint()
+            value = ip_address(bytes(value))
         else:
             value = str(value)
     elif isinstance(value, (ObjectIdentity, univ.ObjectIdentifier)):

src/zino/tasks/bfdtask.py

@@ -4,9 +4,9 @@
 from zino.oid import OID
 from zino.scheduler import get_scheduler
 from zino.snmp import SparseWalkResponse
-from zino.statemodels import BFDEvent, BFDSessState, BFDState, Port
+from zino.statemodels import BFDEvent, BFDSessState, BFDState, IPAddress, Port
 from zino.tasks.task import Task
-from zino.utils import parse_ip, reverse_dns
+from zino.utils import reverse_dns
 
 _log = logging.getLogger(__name__)
 
@@ -142,19 +142,13 @@ async def _get_single_row(self, session_index: int, *variables: Sequence[str]) -
         row = {var.object: val for var, val in response}
         return {OID((session_index,)): row}
 
-    def _parse_row(self, index: OID, state: str, discr: int, addr: str) -> BFDState:
-        try:
-            ipaddr = parse_ip(addr)
-        except ValueError as e:
-            _log.error(f"Error converting addr {addr} to an IP address on device {self.device.name}: {e}")
-            ipaddr = None
-
+    def _parse_row(self, index: OID, state: str, discr: int, addr: IPAddress) -> BFDState:
         # convert from OID object to int
         session_index = int(index[0])
         bfd_state = BFDState(
             session_state=BFDSessState(state),
             session_index=session_index,
             session_discr=discr,
-            session_addr=ipaddr,
+            session_addr=addr,
         )
         return bfd_state

src/zino/tasks/bgpstatemonitortask.py

@@ -13,7 +13,6 @@
     IPAddress,
 )
 from zino.tasks.task import Task
-from zino.utils import parse_ip
 
 _logger = logging.getLogger(__name__)
 
@@ -47,9 +46,9 @@
 }
 BUGGY_REMOTE_ADDRESSES = [
     # Bug in JunOS -- info from IPv6 BGP sessions spill over
-    "0.0.0.0",
+    ip_address("0.0.0.0"),
     # Bug in earlier Cisco IOS, info from elsewhere (IPv6?) spills over
-    "32.1.7.0",
+    ip_address("32.1.7.0"),
 ]
 
 
@@ -159,7 +158,7 @@ async def _get_cisco_bgp_info(self) -> Optional[list[BaseBGPRow]]:
             return None
 
         for oid, result in cisco_bgp_info.items():
-            result["cbgpPeer2RemoteAddr"] = oid
+            result["cbgpPeer2RemoteAddr"] = ip_address(oid)
 
         cisco_bgp_info = self._transform_variables_from_specific_to_general(
             bgp_info=cisco_bgp_info, bgp_style=BGPStyle.CISCO
@@ -227,18 +226,7 @@ def _transform_variables_from_specific_to_general(
             _logger.info(f"router {self.device.name} misses BGP variables ({missing_variables})")
             return None
 
-        results = list()
-
-        # Fix up peer remote address and transform to BaseBGPRow
-        for result in generalized_bgp_info.values():
-            try:
-                result["peer_remote_address"] = parse_ip(result["peer_remote_address"])
-            except ValueError:
-                _logger.debug(f"{self.device_state.name}: Invalid peer_remote_address {result['peer_remote_address']}")
-            else:
-                results.append(BaseBGPRow(**result))
-
-        return results
+        return [BaseBGPRow(**result) for result in generalized_bgp_info.values()]
 
     def _update_single_bgp_entry(self, data: BaseBGPRow, local_as: int, uptime: int):
         if data.peer_remote_address in BUGGY_REMOTE_ADDRESSES:

src/zino/utils.py

@@ -3,36 +3,10 @@
 import os
 import stat
 from functools import wraps
-from ipaddress import ip_address
 from time import time
 from typing import Optional, Union
 
 import aiodns
-from pyasn1.type.univ import OctetString
-
-from zino.statemodels import IPAddress
-
-
-def parse_ip(ip: str) -> IPAddress:
-    """Parses IPv4 and IPv6 addresses in different formats"""
-    try:
-        return ip_address(ip)
-    except ValueError:
-        if ip.startswith("0x"):
-            return _parse_hexa_string_ip(ip)
-        if ":" in ip:
-            return _parse_colon_separated_ip(ip)
-        raise ValueError(f"Input {ip} could not be converted to IP address.")
-
-
-def _parse_hexa_string_ip(ip: str) -> IPAddress:
-    """Parses IP addresses formatted as hexastrings, e.g. 0x7f000001"""
-    return ip_address(bytes(OctetString(hexValue=ip[2:])))
-
-
-def _parse_colon_separated_ip(ip: str) -> IPAddress:
-    """Parses IP addresses formatted with a colon symbol separating every octet, e.g. 7F:00:00:01"""
-    return ip_address(bytes(OctetString(hexValue=ip.replace(":", ""))))
 
 
 async def reverse_dns(ip: str) -> Optional[str]:

tests/tasks/test_bfdtask.py

@@ -16,7 +16,7 @@ def test_parse_row_creates_correct_bfd_state(self, task, bfd_state):
             OID(f".{bfd_state.session_index}"),
             "up",
             bfd_state.session_discr,
-            "0x7f000001",
+            IPv4Address("127.0.0.1"),
         )
         assert state == bfd_state
 

tests/utils_test.py

@@ -1,53 +1,12 @@
 import logging
 import os
 import stat
-from ipaddress import IPv4Address, IPv6Address
 from unittest.mock import AsyncMock, MagicMock
 
 import aiodns
 import pytest
 
-from zino.utils import file_is_world_readable, log_time_spent, parse_ip, reverse_dns
-
-
-class TestParseIP:
-    def test_should_parse_normal_ipv4_string_correctly(self):
-        ip_string = "127.0.0.1"
-        ip = parse_ip(ip_string)
-        assert ip == IPv4Address(ip_string)
-
-    def test_should_parse_normal_ipv6_string_correctly(self):
-        ip_string = "13c7:db1c:4430:c826:6333:aed0:e605:3a3b"
-        ip = parse_ip(ip_string)
-        assert ip == IPv6Address(ip_string)
-
-    def test_should_parse_hexastring_ipv4_correctly(self):
-        ip = parse_ip("0x7f000001")
-        assert ip == IPv4Address("127.0.0.1")
-
-    def test_should_parse_hexastring_ipv6_correctly(self):
-        ip = parse_ip("0x13c7db1c4430c8266333aed0e6053a3b")
-        assert ip == IPv6Address("13c7:db1c:4430:c826:6333:aed0:e605:3a3b")
-
-    def test_should_parse_colon_separated_ipv4_octets_correctly(self):
-        ip = parse_ip("7f:00:00:01")
-        assert ip == IPv4Address("127.0.0.1")
-
-    def test_should_parse_colon_separated_ipv6_octets_correctly(self):
-        ip = parse_ip("13:c7:db:1c:44:30:c8:26:63:33:ae:d0:e6:05:3a:3b")
-        assert ip == IPv6Address("13c7:db1c:4430:c826:6333:aed0:e605:3a3b")
-
-    def test_should_raise_valueerror_if_invalid_ip_format(self):
-        with pytest.raises(ValueError):
-            parse_ip("invalidformat")
-
-    def test_should_raise_error_if_invalid_hexastring(self):
-        with pytest.raises(ValueError):
-            parse_ip("0xinvalidstring")
-
-    def test_should_raise_error_if_invalid_colon_separated_string(self):
-        with pytest.raises(ValueError):
-            parse_ip(":thisis:just:randomstuff")
+from zino.utils import file_is_world_readable, log_time_spent, reverse_dns
 
 
 class TestReverseDNS: