Fix unintended breakdown of MIB object sub-indexes into symbolic representations

changelog.d/389.fixed.md

@@ -0,0 +1 @@
+Fixed unintended symbolic translations in the SNMP abstraction layer, and removed now obsolete workarounds for the problem.

src/zino/oid.py

@@ -35,6 +35,11 @@ def __new__(cls, oid):
             return oid
         return tuple.__new__(cls, oid)
 
+    def __init__(self, oid):
+        super().__init__()
+        if not all(isinstance(i, int) for i in self):
+            raise TypeError("Not all arguments could be converted to `int`")
+
     def __str__(self):
         return SEPARATOR + SEPARATOR.join([str(i) for i in self])
 

src/zino/snmp.py

@@ -499,9 +499,15 @@ def udp_transport_target(self) -> Union[UdpTransportTarget, Udp6TransportTarget]
 
 def _convert_varbind(ident: ObjectIdentity, value: ObjectType) -> SNMPVarBind:
     """Converts a PySNMP varbind pair to an Identifier/value pair"""
-    mib, obj, row_index = ident.getMibSymbol()
+    mib, obj, indices = ident.getMibSymbol()
     value = _mib_value_to_python(value)
-    return Identifier(mib, obj, OID(row_index)), value
+
+    prefix = SNMP._oid_to_object_type(mib, obj)
+    SNMP._resolve_object(prefix)
+    prefix = OID(prefix[0])
+    row_index = OID(ident).strip_prefix(prefix)
+
+    return Identifier(mib, obj, row_index), value
 
 
 def _mib_value_to_python(value: SupportedTypes) -> Union[str, int, OID]:

src/zino/tasks/bgpstatemonitortask.py

@@ -158,7 +158,8 @@ async def _get_cisco_bgp_info(self) -> Optional[list[BaseBGPRow]]:
             return None
 
         for oid, result in cisco_bgp_info.items():
-            result["cbgpPeer2RemoteAddr"] = ip_address(oid)
+            _addr_type, _addr_len, addr = oid[0], oid[1], oid[2:]
+            result["cbgpPeer2RemoteAddr"] = ip_address(bytes(addr))
 
         cisco_bgp_info = self._transform_variables_from_specific_to_general(
             bgp_info=cisco_bgp_info, bgp_style=BGPStyle.CISCO
@@ -191,16 +192,7 @@ async def _get_bgp_info(self, mib_name: str, variables: Iterable[str]) -> Option
             *((mib_name, var) for var in variables),
             max_repetitions=3,
         )
-
-        cleaned_bgp_info = dict()
-
-        for oid, entry in bgp_info.items():
-            if len(oid) == 1:
-                cleaned_bgp_info[oid[0].prettyPrint()] = entry
-            else:
-                cleaned_bgp_info[oid[1].prettyPrint()] = entry
-
-        return cleaned_bgp_info
+        return bgp_info
 
     def _transform_variables_from_specific_to_general(
         self, bgp_info: SparseWalkResponse, bgp_style: BGPStyle

tests/oid_test.py

@@ -1,3 +1,5 @@
+import pytest
+
 from zino.oid import OID
 
 
@@ -38,3 +40,8 @@ def test_can_create_oid_from_bytestring():
     oid_bytestring = b".1.2.3.4"
     oid = OID(oid_bytestring)
     assert str(oid) == ".1.2.3.4"
+
+
+def test_when_oid_contains_non_numeric_elements_it_should_raise_typeerror():
+    with pytest.raises(TypeError):
+        OID((1, 2, 3, "foo", 4))