Use SNMPv3 contexts for collecting logical BRIDGE-MIB instances (primarily Cisco)

python/nav/ipdevpoll/snmp/common.py

@@ -157,6 +157,8 @@ class SNMPParameters:
     auth_password: Optional[str] = None
     priv_protocol: Optional[PrivacyProtocol] = None
     priv_password: Optional[str] = None
+    context_name: Optional[str] = None
+    context_engine_id: Optional[str] = None
 
     # Specific to ipdevpoll-derived implementations
     throttle_delay: int = 0
@@ -254,6 +256,10 @@ def as_agentproxy_args(self) -> dict[str, Any]:
                 params.extend(["-x", self.priv_protocol.value])
             if self.priv_password:
                 params.extend(["-X", self.priv_password])
+            if self.context_name:
+                params.extend(["-n", self.context_name])
+            if self.context_engine_id:
+                params.extend(["-E", self.context_engine_id])
             kwargs["cmdLineArgs"] = tuple(params)
 
         return kwargs

python/nav/ipdevpoll/utils.py

@@ -17,6 +17,8 @@
 #
 """Utility functions for ipdevpoll."""
 
+from __future__ import annotations
+
 import logging
 import re
 from typing import TYPE_CHECKING
@@ -33,6 +35,7 @@
 
 if TYPE_CHECKING:
     from nav.mibs.bridge_mib import MultiBridgeMib
+    from nav.mibs.types import LogicalMibInstance
 
 _logger = logging.getLogger(__name__)
 MAX_MAC_ADDRESS_LENGTH = 6
@@ -144,15 +147,12 @@ async def get_multibridgemib(agentproxy) -> "MultiBridgeMib":
     return MultiBridgeMib(agentproxy, instances)
 
 
-async def get_dot1d_instances(agentproxy):
+async def get_dot1d_instances(agentproxy) -> "list[LogicalMibInstance]":
     """
     Gets a list of alternative BRIDGE-MIB instances from a Cisco or Aruba
     agent.
 
-    First
-
-    :returns: A list of [(description, community), ...] for each alternate
-              BRIDGE-MIB instance.
+    :returns: A list of LogicalMibInstance for each alternate BRIDGE-MIB instance.
 
     """
     from nav.mibs.snmpv2_mib import Snmpv2Mib
@@ -201,8 +201,12 @@ def get_arista_vrf_instances(agentproxy) -> Deferred:
 
     vrf_mib = AristaVrfMib(agentproxy)
     states = yield vrf_mib.get_vrf_states(only='active')
-    vrfs = [('', agentproxy.community)]
-    vrfs.extend((vrf, f"{agentproxy.community}@{vrf}") for vrf in states)
+    # XXX: This part does not currently support SNMPv3, as we have no known way to
+    #      derive the correct SNMPv3 context name for each VRF.
+    vrfs = [LogicalMibInstance('', agentproxy.community)]
+    vrfs.extend(
+        LogicalMibInstance(vrf, f"{agentproxy.community}@{vrf}") for vrf in states
+    )
     return vrfs
 
 

python/nav/mibs/cisco_vtp_mib.py

@@ -18,6 +18,7 @@
 from nav.bitvector import BitVector
 from nav.smidumps import get_mib
 from . import mibretriever
+from .types import LogicalMibInstance
 
 CHARS_IN_1024_BITS = 128
 
@@ -98,13 +99,14 @@ def get_operational_vlans(self):
         states = yield self.get_ethernet_vlan_states()
         return set(vlan for vlan, state in states.items() if state == 'operational')
 
-    async def retrieve_alternate_bridge_mibs(self):
+    async def retrieve_alternate_bridge_mibs(self) -> list[LogicalMibInstance]:
         """Retrieve a list of alternate bridge mib instances.
 
-        :returns: A list of (descr, community) tuples for each operational
-                  VLAN on this device.
+        :returns: A list of LogicalMibInstance for each operational VLAN on this device.
 
         """
         vlans = await self.get_operational_vlans()
         community = self.agent_proxy.community
-        return [("vlan%s" % vlan, "%s@%s" % (community, vlan)) for vlan in vlans]
+        return [
+            LogicalMibInstance(f"vlan{vlan}", f"{community}@{vlan}") for vlan in vlans
+        ]

python/nav/mibs/entity_mib.py

@@ -27,6 +27,7 @@
 from nav.oids import OID
 from nav.smidumps import get_mib
 from nav.mibs import mibretriever
+from nav.mibs.types import LogicalMibInstance
 from nav.ipdevpoll.shadows import PowerSupplyOrFan, Device
 
 _logger = logging.getLogger(__name__)
@@ -37,13 +38,10 @@ class EntityMib(mibretriever.MibRetriever):
 
     mib = get_mib('ENTITY-MIB')
 
-    async def retrieve_alternate_bridge_mibs(self):
+    async def retrieve_alternate_bridge_mibs(self) -> list[LogicalMibInstance]:
         """Retrieves a list of alternate bridge mib instances.
 
-        This is accomplished by looking at entLogicalTable.  Returns a
-        list of tuples::
-
-          (entity_description, community)
+        This is accomplished by looking at entLogicalTable.
 
         :NOTE: Some devices will return entities with the same community.
                These should effectively be filtered out for polling purposes.
@@ -63,10 +61,21 @@ def _is_bridge_mib_instance_with_valid_community(row):
             )
 
         result = await self.retrieve_columns(
-            ['entLogicalDescr', 'entLogicalType', 'entLogicalCommunity']
+            [
+                'entLogicalDescr',
+                'entLogicalType',
+                'entLogicalCommunity',
+                'entLogicalContextEngineID',
+                'entLogicalContextName',
+            ]
         )
         return [
-            (r["entLogicalDescr"], r["entLogicalCommunity"].decode("utf-8"))
+            LogicalMibInstance(
+                r["entLogicalDescr"],
+                r["entLogicalCommunity"].decode("utf-8"),
+                r["entLogicalContextName"],
+                r["entLogicalContextEngineID"],
+            )
             for r in result.values()
             if _is_bridge_mib_instance_with_valid_community(r)
         ]

python/nav/mibs/mibretriever.py

@@ -31,8 +31,9 @@
 
 """
 
+import dataclasses
 import logging
-from typing import Awaitable
+from typing import Awaitable, Iterator
 
 from pynetsnmp.netsnmp import SnmpTimeoutError
 from twisted.internet import defer, reactor
@@ -43,6 +44,7 @@
 from nav.ipdevpoll import ContextLogger
 from nav.ipdevpoll.utils import fire_eventually
 from nav.errors import GeneralException
+from nav.mibs.types import LogicalMibInstance
 from nav.oids import OID
 from nav.smidumps import get_mib
 
@@ -577,38 +579,34 @@ def retrieve_column_by_index(self, column, index):
 
 
 class MultiMibMixIn(MibRetriever):
-    """Queries and chains the results of multiple MIB instances using
-    community indexing.
+    """Queries and chains the results of multiple logical MIB instances, using
+    either community indexing or SNMPv3 contexts.
 
     Useful for Cisco devices, whose SNMP agents employ multiple BRIDGE-MIB
-    instances, one for each active VLAN, each indexable via a modified SNMP
-    community.
+    instances, one for each active VLAN.
 
     Add the mixin to the list of base classes of a MibRetriever descendant
     class, and override any querying method that should work across multiple
     instances.  The overriden method should use a call to self._multiquery().
 
     """
 
-    def __init__(self, agent_proxy, instances):
+    def __init__(self, agent_proxy, instances: list[LogicalMibInstance]):
         """Initializes a MultiBridgeQuery to perform SNMP requests on multiple
-        BRIDGE-MIB instances on the same host/IP.
+        logical MIB instances on the same host/IP.
 
         :param agent_proxy: The base AgentProxy to use for communication. An
-                            AgentProxy for each additional BRIDGE-MIB instance
+                            AgentProxy for each additional MIB instance
                             will be created based on the properties of this
                             one.
 
-        :param instances: A sequence of tuples describing the MIB instances to
-                          query, like [(description, community), ...], where
-                          description is any object that can be used to
-                          identify an instance, and community is the alternate
-                          MIB instance's SNMP read-community.
+        :param instances: A list of LogicalMibInstance objects that describe the
+                          logical instances to query.
 
         """
         super(MultiMibMixIn, self).__init__(agent_proxy)
         self._base_agent = agent_proxy
-        self.instances = instances
+        self.instances: list[LogicalMibInstance] = instances
 
     @defer.inlineCallbacks
     def _multiquery(self, method, *args, **kwargs):
@@ -651,6 +649,8 @@ class constructor; result is the actual result
                 if agent is not self._base_agent:
                     agent.close()
                 self.agent_proxy = self._base_agent
+            if agent is not self._base_agent:
+                self._logger.debug("got result from %r: %r", descr, one_result)
             results.append((descr, one_result))
             yield lambda thing: fire_eventually(thing)
         return integrator(results)
@@ -687,46 +687,53 @@ def _dictintegrator(results):
         return merged_dict
 
     def _make_agents(self):
-        "Generates a series of alternate AgentProxy instances"
+        """Generates a series of alternate AgentProxy instances"""
         instances = list(self._prune_instances())
         if not instances:
             # The un-indexed BRIDGE-MIB instance represents the default
             # VLAN. We only check this un-indexed instance if no alternate
             # instances were found, otherwise some results will be duplicated.
-            yield (self._base_agent, None)
-        for descr, community in instances:
-            agent = self._get_alternate_agent(community)
-            yield (agent, descr)
+            yield self._base_agent, None
+        for instance in instances:
+            agent = self._get_alternate_agent(instance)
+            yield agent, instance.description
 
-    def _prune_instances(self):
-        """ "Prunes instances with duplicate community strings from the
+    def _prune_instances(self) -> Iterator[LogicalMibInstance]:
+        """Prunes instances with duplicate community strings from the
         instance list, as these cannot possibly represent individual MIB
         instances in the queried devices.
 
         """
         seen_communities = set(self._base_agent.community)
 
-        for descr, community in self.instances:
-            if community not in seen_communities:
-                seen_communities.add(community)
-                yield (descr, community)
+        for instance in self.instances:
+            if not instance.community:
+                continue
+            if instance.community not in seen_communities:
+                seen_communities.add(instance.community)
+                yield instance
 
-    def _get_alternate_agent(self, community):
-        """Create an alternate AgentProxy using a different community.
+    def _get_alternate_agent(self, instance: LogicalMibInstance):
+        """Create an alternate AgentProxy using the settings for a logical MIB instance.
 
         :returns: An instance of the same class as the AgentProxy object given
                   to __init__().  Every main attribute will be copied from the
-                  original AgentProxy, except for the community string, which
-                  will be taken from the MIB instance list..
-
+                  original AgentProxy, except for the alternative attributes
+                  described by a LogicalMibInstance object.
         """
         agent = self._base_agent
+        if agent.snmp_parameters.version == 3:
+            changes = {"context_name": instance.context}
+            if instance.context_engine_id:
+                changes["context_engine_id"] = instance.context_engine_id.hex()
+        else:
+            changes = {"community": instance.community}
+        new_parameters = dataclasses.replace(agent.snmp_parameters, **changes)
+
         alt_agent = agent.__class__(
             agent.ip,
             agent.port,
-            community=community,
-            snmpVersion=agent.snmpVersion,
-            snmp_parameters=agent.snmp_parameters,
+            snmp_parameters=new_parameters,
         )
         if hasattr(agent, 'protocol'):
             alt_agent.protocol = agent.protocol

python/nav/mibs/types.py

@@ -0,0 +1,32 @@
+#
+# Copyright (C) 2025 Sikt
+#
+# This file is part of Network Administration Visualized (NAV).
+#
+# NAV is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License version 3 as published by
+# the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+# more details.  You should have received a copy of the GNU General Public
+# License along with NAV. If not, see <http://www.gnu.org/licenses/>.
+#
+"""Type definitions for nav.mibs package."""
+
+from typing import NamedTuple, Optional
+
+
+class LogicalMibInstance(NamedTuple):
+    """A MIB instance identifier.
+
+    Some devices (like Cisco switches) provide multiple logical instances of a MIB (such
+    as BRIDGE-MIB) within a single physical device, and we need a way to reference them
+    and the settings needed to collect data from a specific logical instance.
+    """
+
+    description: str
+    community: Optional[str]
+    context: Optional[str] = None
+    context_engine_id: Optional[bytes] = None